如何基于Kafka Streams的GlobalKTable实现反向查找？

背景 #

我有一定Kafka使用经验，但对Kafka Streams API及流处理概念较为陌生。目前需要构建一个Kafka流应用，通过额外的Kafka主题提供的查找表实现编解码：

• 业务主题均为共分区，包含5个分区；
• Lookup mapping流是变更数据流，已做压缩处理，且仅含1个分区；
• 编码规则采用ASCII（如'a'→97），Lookup mapping主题以字母为键，负载同时包含字母与对应的编码值。

已实现的编码逻辑 #

通过GlobalKTable与LeftJoin实现了编码功能，代码如下：

final var mappingTable = streamsBuilder.globalTable(LOOKUP_MAPPINGS, ...);

streamsBuilder.stream(PLAIN_REQUESTS, ...)
  .leftJoin(mappingTable, (key, value) -> value.getPlain(), EncoderHelper::encode)
  .to(ENCODED_REQUESTS, ...);

遇到的问题 #

编码功能正常，但解码时需基于编码值（如97）查找对应明文，而GlobalKTable的键为明文，无法直接实现反向查找。

已排除的方案 #

• 通过Kafka Consumer API手动存储数据并编解码：可行但需确保数据加载完成后再处理，存在重复造轮子问题；
• 外部创建以编码值为键的主题：会导致两个主题强耦合，设计不合理；
• 程序内重分区后生成GlobalKTable：会重复存储映射，增加内存占用，不符合GlobalKTable的预期用法；
• 重分区后使用KTable：因共分区要求复杂度高，设计不合理。

需求 #

寻求符合Kafka Streams API设计原则的双向查找系统实现思路，向有经验的流处理开发者请教解决方案。

解决方案思路 #

1. 自定义双向状态存储 #

在构建GlobalKTable时，不使用默认状态存储，而是自定义一个同时维护正向（明文→编码）和反向（编码→明文）映射的KeyValueStore，直接从状态存储中支持双向查询。

示例代码思路：

// 自定义双向状态存储
public class BidirectionalStore implements KeyValueStore<String, EncodingValue> {
    private final KeyValueStore<String, EncodingValue> forwardStore;
    private final KeyValueStore<Integer, String> reverseStore;

    public BidirectionalStore(KeyValueStore<String, EncodingValue> forwardStore, KeyValueStore<Integer, String> reverseStore) {
        this.forwardStore = forwardStore;
        this.reverseStore = reverseStore;
    }

    @Override
    public void put(String key, EncodingValue value) {
        forwardStore.put(key, value);
        reverseStore.put(value.getCode(), key);
    }

    // 实现其他KeyValueStore接口方法，同步维护两个存储
    // 添加自定义方法支持反向查找
    public String getByCode(Integer code) {
        return reverseStore.get(code);
    }
}

// 构建自定义存储的StoreBuilder
final StoreBuilder<BidirectionalStore> bidirectionalStoreBuilder = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("bidirectional-mapping"),
    Serdes.String(),
    encodingValueSerde
).withCachingEnabled();

// 添加全局存储并绑定Lookup mapping主题
streamsBuilder.addGlobalStore(
    bidirectionalStoreBuilder,
    LOOKUP_MAPPINGS,
    Consumed.with(Serdes.String(), encodingValueSerde),
    () -> new Processor<String, EncodingValue>() {
        private BidirectionalStore store;

        @Override
        public void init(ProcessorContext context) {
            store = context.getStateStore("bidirectional-mapping");
        }

        @Override
        public void process(String key, EncodingValue value) {
            store.put(key, value);
        }

        @Override
        public void close() {}
    }
);

// 解码逻辑：从自定义存储中查询明文
streamsBuilder.stream(ENCODED_REQUESTS, ...)
    .transformValues(() -> new ValueTransformerWithKey<String, EncodedValue, String>() {
        private BidirectionalStore store;

        @Override
        public void init(ProcessorContext context) {
            store = context.getStateStore("bidirectional-mapping");
        }

        @Override
        public String transform(String key, EncodedValue value) {
            return store.getByCode(value.getCode());
        }

        @Override
        public void close() {}
    })
    .to(PLAIN_RESPONSES, ...);

2. 内部维护反向GlobalKTable #

将原Lookup mapping流转换为以编码值为键的内部流，构建一个反向GlobalKTable用于解码。内部主题由Kafka Streams自动管理，避免外部强耦合，且因为原主题是压缩状态，内存占用可控。

示例代码：

// 正向GlobalKTable用于编码
final var forwardMapping = streamsBuilder.globalTable(LOOKUP_MAPPINGS, Consumed.with(Serdes.String(), encodingValueSerde));

// 将Lookup mapping流转换为编码值为键的内部流，生成反向GlobalKTable
streamsBuilder.stream(LOOKUP_MAPPINGS, Consumed.with(Serdes.String(), encodingValueSerde))
    .selectKey((k, v) -> v.getCode())
    .to("reverse-lookup-mappings", Produced.with(Serdes.Integer(), Serdes.String()));

final var reverseMapping = streamsBuilder.globalTable("reverse-lookup-mappings", Consumed.with(Serdes.Integer(), Serdes.String()));

// 编码逻辑保持不变
streamsBuilder.stream(PLAIN_REQUESTS, ...)
    .leftJoin(forwardMapping, (key, value) -> value.getPlain(), EncoderHelper::encode)
    .to(ENCODED_REQUESTS, ...);

// 解码逻辑使用反向GlobalKTable
streamsBuilder.stream(ENCODED_REQUESTS, ...)
    .leftJoin(reverseMapping, (key, value) -> value.getCode(), (encodedValue, plainText) -> plainText)
    .to(PLAIN_RESPONSES, ...);

方案对比 #

• 自定义状态存储：内存占用更低（仅存一份数据维护双向映射），但需要自定义存储和处理器，代码复杂度稍高；
• 反向GlobalKTable：代码简洁，符合Kafka Streams常规用法，内存占用略高但可控，适合数据量不大的场景。

内容的提问来源于stack exchange，提问作者simon