本文介紹了無法打開Kafka流的存儲，因為狀態無效的處理方法，對大家解決問題具有一定的參考價值，需要的朋友們下面隨著小編來一起學習吧！
問題描述
我正在嘗試使用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狀態下完成初始化后調用了createFlowHistory，但我始終無法執行此操作，KafkaStreams報告以下錯誤：
線程Main中出現異常
Org.apache.kafka.streams.errors.InvalidStateStoreException：無法獲取
狀態存儲flow-event-stream-file-service-test-instance-by-key
因為流線程是PARTIONS_ASSIGNED，而不是運行
顯然，線程的狀態已更改。在嘗試查詢存儲區并等待Kafka的內部線程進入正確狀態時，是否需要手動處理此問題？
推薦答案
舊版本(2.2.0之前的)
啟動時，Kafka Streams執行以下狀態轉換：
CREATED -> RUNNING -> REBALANCING -> RUNNING
您需要等待第二個運行狀態才能進行查詢。
新版本：2.2.0
啟動時的狀態轉換行為(通過https://issues.apache.org/jira/browse/KAFKA-7657)更改為：
CREATED -> REBALANCING -> RUNNING
因此，您不應再遇到此問題。
這篇關于無法打開Kafka流的存儲，因為狀態無效的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，
