无法为 Kafka 流打开存储，因为状态无效

Question

我正在尝试使用 Kafka Streams 并创建了以下拓扑：

    KStream<String, HistoryEvent> eventStream = builder.stream(applicationTopicName, Consumed.with(Serdes.String(),
            historyEventSerde));

    eventStream.selectKey((key, value) -> new HistoryEventKey(key, value.getIdentifier()))
            .groupByKey()
            .reduce((e1, e2) -> e2, Materialized.as(streamByKeyStoreName));

稍后我会这样开始直播：

private void startKafkaStreams(KafkaStreams streams) {
    CompletableFuture<KafkaStreams.State> stateFuture = new CompletableFuture<>();
    streams.setStateListener((newState, oldState) -> {
        if(stateFuture.isDone()) {
            return;
        }

        if(newState == KafkaStreams.State.RUNNING || newState == KafkaStreams.State.ERROR) {
            stateFuture.complete(newState);
        }
    });

    streams.start();
    try {
        KafkaStreams.State finalState = stateFuture.get();
        if(finalState != KafkaStreams.State.RUNNING) {
            // ...
        }
    } catch (InterruptedException ex) {
        // ...
    } catch(ExecutionException ex) {
        // ...
    }
}

我的流开始时没有错误，它们最终进入 RUNNING 状态，未来已完成。稍后我尝试访问我在拓扑中为 KTable 创建的存储：

public KafkaFlowHistory createFlowHistory(String flowId) {
    ReadOnlyKeyValueStore<HistoryEventKey, HistoryEvent> store = streams.store(streamByKeyStoreName,
            QueryableStoreTypes.keyValueStore());
    return new KafkaFlowHistory(flowId, store, event -> topicProducer.send(new ProducerRecord<>(applicationTopicName, flowId, event)));
}

我已经验证 createFlowHistory 在 RUNNING 状态下完成初始化 future 后被调用，但是我一直无法执行此操作并且 KafkaStreams 报告以下错误：

Exception in thread "main" org.apache.kafka.streams.errors.InvalidStateStoreException: Cannot get state store flow-event-stream-file-service-test-instance-by-key because the stream thread is PARTITIONS_ASSIGNED, not RUNNING

显然线程的状态已经改变。在尝试查询商店并等待 Kafka 的内部线程进入正确状态时，我是否需要手动处理此问题？

Answer 1

旧版本（之前 2.2.0）

启动时，Kafka Streams 会进行以下状态转换：

CREATED -> RUNNING -> REBALANCING -> RUNNING

需要等待第二个运行状态才能查询。

新版本： 截至 2.2.0

启动时的状态转换行为已更改（通过 https://issues.apache.org/jira/browse/KAFKA-7657）为：

CREATED -> REBALANCING -> RUNNING

所以，你应该不会再遇到这个问题了。

无法为 Kafka 流打开存储，因为状态无效

Unable to open store for Kafka streams because invalid state

java

apache-kafka

apache-kafka-streams