Why am I occasionally getting a InvalidStateStoreException PARTITIONS_REVOKED, not RUNNING when retrieving a store to query it?

user_1729795 0
Viewed 2355

I am accessing a state store to query it and have had to wrap the store() statement with a try/catch block to retry it because sometimes I am getting this exception:

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)

This is the code used to retrieve the store (the full code is on a 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
}

And this is the main program:

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()
}

The exception happens randomly running the program several times, after the previous executions finish. Note: I don't do anything to the executing Kafka cluster and use its default config.

apache-kafka-streams apache-kafka
1 Answers

At the time you are accessing the store, the Kafka Streams application is going through a rebalance, and state stores aren't accessible at that time. You want to make sure you only query the stores when the application state is RUNNING and not REBALANCING.

What you could do is check the state of the application before attempting to read from the store like this:

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

There is also a KafkaStreams.setStateListener method you can use to register a KafkStreams.StateListener implementation. The StateListener.onChange method is called each time the application changes its state.

user_18200130
Related