我正在尝试使用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 但是，我始终无法做到这一点，kafkastreams报告了以下错误：
线程“main”org.apache.kafka.streams.errors.invalidstatestoreexception中出现异常：无法按键获取状态存储流事件流文件服务测试示例，因为流线程已分配，未运行
显然线程的状态已经改变了。当我试图查询一个存储并等待kafka的内部线程进入正确的状态时，是否需要手动处理这个问题？