我正在尝试使用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) {
// ...
}
}
我的Streams开始没有错误,他们最终进入 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:无法获取状态存储流 - 事件 - 流 - 文件 - 服务 - 测试 - 实例 - 因为流线程是PARTITIONS_ASSIGNED,而不是RUNNING
显然线程的状态已经改变 . 在尝试查询商店并等待Kafka的内部线程进入正确状态时,是否需要手动处理此问题?
1 回答
在启动时,Kafka Streams执行以下状态转换:
您需要等待第二个RUNNING状态才能进行查询 . 这是一个已知问题,我们希望在将来的版本中修复它 .