Guozhang Wang created KAFKA-7112: ------------------------------------ Summary: StreamThread does not check for state again after pollRequests() Key: KAFKA-7112 URL: https://issues.apache.org/jira/browse/KAFKA-7112 Project: Kafka Issue Type: Bug Components: streams Reporter: Guozhang Wang
In StreamThread's main loop, we have: {code} if (state == State.PARTITIONS_ASSIGNED) { // try to fetch some records with zero poll millis // to unblock the restoration as soon as possible records = pollRequests(Duration.ZERO); if (taskManager.updateNewAndRestoringTasks()) { setState(State.RUNNING); } } {code} in which we first check for state, and if it is {{PARTITIONS_ASSIGNED}} then call `consumer.poll()` and then call `askManager.updateNewAndRestoringTasks()`. There is a race condition though, that if another rebalance gets triggered, then `onPartitionRevoked` will be called in which we will {{restoreConsumer.unsubscribe();}}, and then if we call {{taskManager.updateNewAndRestoringTasks()}} right away we will see this: {code} Encountered the following error during processing: (org.apache.kafka.streams.processor.internals.StreamThread) java.lang.IllegalStateException: Consumer is not subscribed to any topics or assigned any partitions at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1159) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1150) at org.apache.kafka.streams.processor.internals.StoreChangelogReader.restore(StoreChangelogReader.java:83) at org.apache.kafka.streams.processor.internals.TaskManager.updateNewAndRestoringTasks(TaskManager.java:317) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:808) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736) {code} -- This message was sent by Atlassian JIRA (v7.6.3#76005)