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

Unable to open store for Kafka streams because invalid state(无法打开Kafka流的存储,因为状态无效)

本文介绍了无法打开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)));
}

我已验证createFlowHistoryRUNNING状态下完成初始化后调用了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流的存储,因为状态无效的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持编程学习网!

本文标题为:无法打开Kafka流的存储,因为状态无效

基础教程推荐