KStream重新分区后线程分配异常，状态存储无法找到对应Key

问题背景 #

需要消费两个Topic：

• Topic A：含40个分区，消息Key为字符串类型的社会保障号（SSN），消费后将条目存入以SSN为Key、自定义Java类为Value的共享持久化状态存储。
• Topic B：含10个分区，消息Key不是SSN，但可从消息内容提取SSN。执行selectKey和repartition操作后，尝试从状态存储获取对应SSN条目时返回null。

添加日志后发现：处理同一SSN时，Topic A流与Topic B流的taskId、线程名称不同，导致状态存储中无法找到对应Key。

问题原因 #

Kafka Streams中状态存储与任务（Task）绑定，每个任务对应一组分区，状态存储数据按Key哈希值分配到对应任务分区。你的代码存在两个核心问题：

1. 分区器不一致：
   • Topic A使用默认DefaultPartitioner，基于原始Key（SSN）哈希分区。
   • Topic B重新分区时未指定与Topic A一致的分区器，导致相同SSN的消息重新分区后，分配的任务分区与Topic A中该SSN所在分区不匹配。
2. 状态分片不共享：虽然两个流处理器关联了同一个状态存储名称，但任务分区不对应时，每个任务持有的是独立的状态分片。Topic A处理器在某任务分片写入数据，Topic B处理器在另一任务分片查询，自然找不到数据。

解决方案 #

要让相同SSN的数据落到同一任务的状态存储分片，需满足：

• 两个流的分区数一致（已满足，均为40）。
• 两个流使用完全相同的分区器，且基于相同Key（SSN）分区。

修正后的代码 #

拓扑构建代码 #

// 添加状态存储
var storeBuilder = Stores.keyValueStoreBuilder(
                    Stores.persistentKeyValueStore("stateStoreName"),
                    Serdes.String(),
                    employeeInfoSerde);
streamsBuilder.addStateStore(storeBuilder);

// Topic A 处理流
streamsBuilder.stream("topicA", Consumed.with(Serdes.String(), topicAValueSerde))
    .processValues(() -> new TopicAMessageProcessor("stateStoreName"), "stateStoreName");

// Topic B 处理流
var repartitioned = Repartitioned.with(Serdes.String(), topicBValueSerde)
                .withName("topicB-rekey")
                .withNumberOfPartitions(40)
                .withPartitioner(new DefaultPartitioner()); // 统一分区器，与Topic A保持一致

streamsBuilder.stream("topicB", Consumed.with(Serdes.String(), topicBValueSerde))
    .selectKey((key, value) -> value.getSSN()) // 修正：从消息Value中提取SSN（原代码key.getSSN()应为笔误）
    .repartition(repartitioned)
    .processValues(() -> new TopicBMessageProcessor("stateStoreName"), "stateStoreName");

streamsBuilder.build();

处理器代码（修正类名笔误） #

Topic A 处理器

public class TopicAMessageProcessor extends ContextualFixedKeyProcessor<String, TopicAValue, TopicAValue> {

    private final String storeName;
    private KeyValueStore<String, EmployeeInfo> store;

    public TopicAMessageProcessor(String storeName) {
        super();
        this.storeName = storeName;
    }

    @Override
    public void init(FixedKeyProcessorContext<String, TopicAValue> context) {
        super.init(context);
        store = context.getStateStore(storeName);
    }

    @Override
    public void process(FixedKeyRecord<String, TopicAValue> fixedKeyRecord) {
        String ssn = fixedKeyRecord.key();
        store.put(ssn, new EmployeeInfo(ssn));
    }

}

Topic B 处理器

public class TopicBMessageProcessor extends ContextualFixedKeyProcessor<String, TopicBValue, TopicBValue> {

    private final String storeName;
    private KeyValueStore<String, EmployeeInfo> store;

    public TopicBMessageProcessor(String storeName) {
        super();
        this.storeName = storeName;
    }

    @Override
    public void init(FixedKeyProcessorContext<String, TopicBValue> context) {
        super.init(context);
        store = context.getStateStore(storeName);
    }

    @Override
    public void process(FixedKeyRecord<String, TopicBValue> fixedKeyRecord) {
        String ssn = fixedKeyRecord.key();
        EmployeeInfo info = store.get(ssn); // 现在可以正确获取数据
    }

}

验证要点 #

• 启动后检查日志，确认同一SSN的消息在两个流中对应相同的taskId。
• 可通过Kafka Streams状态查询工具验证数据写入和读取是否正常。

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