为什么我在检索存储以查询它时偶尔会收到 InvalidStateStoreException PARTITIONS_REVOKED，而不是 RUNNING？答案

【问题标题】：Why am I occasionally getting a InvalidStateStoreException PARTITIONS_REVOKED, not RUNNING when retrieving a store to query it?为什么我在检索存储以查询它时偶尔会收到 InvalidStateStoreException PARTITIONS_REVOKED，而不是 RUNNING？
【发布时间】：2018-08-30 22:30:03
【问题描述】：

我正在访问一个状态存储来查询它，并且不得不用 try/catch 块包装 store() 语句来重试它，因为有时我会遇到这个异常：

org.apache.kafka.streams.errors.InvalidStateStoreException: Cannot get state store customers-store because the stream thread is PARTITIONS_REVOKED, not RUNNING
    at org.apache.kafka.streams.state.internals.StreamThreadStateStoreProvider.stores(StreamThreadStateStoreProvider.java:49)
    at org.apache.kafka.streams.state.internals.QueryableStoreProvider.getStore(QueryableStoreProvider.java:57)
    at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1053)
    at com.codependent.kafkastreams.customer.service.CustomerService.getCustomer(CustomerService.kt:75)
    at com.codependent.kafkastreams.customer.service.CustomerServiceKt.main(CustomerService.kt:108)

这是用于检索商店的代码（完整代码在 github repo 上）：

fun getCustomer(id: String): Customer? {
    var keyValueStore: ReadOnlyKeyValueStore<String, Customer>? = null
    while(keyValueStore == null) {
        try {
            keyValueStore = streams.store(CUSTOMERS_STORE, QueryableStoreTypes.keyValueStore<String, Customer>())
        } catch (ex: InvalidStateStoreException) {
            ex.printStackTrace()
        }
    }
    val customer = keyValueStore.get(id)
    return customer
}

这是主程序：

fun main(args: Array<String>) {
    val customerService = CustomerService("main", "localhost:9092")
    customerService.initializeStreams()
    customerService.createCustomer(Customer("53", "Joey"))
    val customer = customerService.getCustomer("53")
    println(customer)
    customerService.stopStreams()
}

在之前的执行完成后，该异常会随机运行几次。注意：我不对正在执行的 Kafka 集群做任何事情，而是使用它的默认配置。

【问题讨论】：

标签： apache-kafka apache-kafka-streams

【解决方案1】：

在您访问存储时，Kafka Streams 应用程序正在重新平衡，此时无法访问状态存储。您要确保仅在应用程序状态为 RUNNING 而不是 REBALANCING 时查询存储。

您可以做的是在尝试从存储中读取之前检查应用程序的状态，如下所示：

if(streams.state() == State.RUNNING) {
    keyValueStore = streams.store(...);
    val customer = keyValueStore.get(id);
    return customer;
}

还有一个KafkaStreams.setStateListener 方法可用于注册KafkStreams.StateListener 实现。每次应用程序更改其状态时，都会调用 StateListener.onChange 方法。

【讨论】：