[jira] [Created] (KAFKA-8062) StateListener is not notified when StreamThread dies

["Andrey Volkov (JIRA)" <ji...@apache.org>]

Andrey Volkov created KAFKA-8062:
------------------------------------

             Summary: StateListener is not notified when StreamThread dies
                 Key: KAFKA-8062
                 URL: https://issues.apache.org/jira/browse/KAFKA-8062
             Project: Kafka
          Issue Type: Bug
          Components: streams
    Affects Versions: 2.1.1
         Environment: Kafka 2.1.1, kafka-streams-scala version 2.1.1
            Reporter: Andrey Volkov


I want my application to react when streams die. Trying to use KafkaStreams.setStateListener. Also checking KafkaStreams.state() from time to time.

The test scenario: Kafka is available, but there are no topics that my Topology is supposed to use.

I expect streams to crash and the state listener to be notified about that, with the new state ERROR. KafkaStreams.state() should also return ERROR.

In reality the streams crash, but the KafkaStreams.state() method always returns REBALANCING and the last time the StateListener was called, the new state was also REBALANCING. 

 

I believe the reason for this is in the methods:

org.apache.kafka.streams.KafkaStreams.StreamStateListener.onChange() which does not react on the state StreamsThread.State.PENDING_SHUTDOWN

and

org.apache.kafka.streams.processor.internals.StreamThread.RebalanceListener.onPartitionsAssigned, which calls shutdown() setting the state to PENDING_SHUTDOWN and then

streamThread.setStateListener(null) effectively removing the state listener, so that the DEAD state of the thread never reaches KafkaStreams object.

Here is an extract from the logs:

{{14:57:03.272 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] ERROR o.a.k.s.p.i.StreamsPartitionAssignor - stream-thread [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1-consumer] test-input-topic is unknown yet during rebalance, please make sure they have been pre-created before starting the Streams application.}}
{{14:57:03.283 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.c.c.i.AbstractCoordinator - [Consumer clientId=Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1-consumer, groupId=Test] Successfully joined group with generation 1}}
{{14:57:03.284 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.c.c.i.ConsumerCoordinator - [Consumer clientId=Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1-consumer, groupId=Test] Setting newly assigned partitions []}}
{{14:57:03.285 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.s.p.i.StreamThread - stream-thread [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] Informed to shut down}}
{{14:57:03.285 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.s.p.i.StreamThread - stream-thread [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] State transition from PARTITIONS_REVOKED to PENDING_SHUTDOWN}}
{{14:57:03.316 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.s.p.i.StreamThread - stream-thread [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] Shutting down}}
{{14:57:03.317 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.c.c.KafkaConsumer - [Consumer clientId=Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1-restore-consumer, groupId=] Unsubscribed all topics or patterns and assigned partitions}}
{{14:57:03.317 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.c.p.KafkaProducer - [Producer clientId=Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1-producer] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.}}
{{14:57:03.332 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.s.p.i.StreamThread - stream-thread [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD}}
{{14:57:03.332 [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] INFO o.a.k.s.p.i.StreamThread - stream-thread [Test-749d56c7-505d-46a4-8dbd-2c756b353adc-StreamThread-1] Shutdown complete}}

After this calls to KafkaStreams.state() still return REBALANCING

There is a workaround with requesting KafkaStreams.localThreadsMetadata() and checking each thread's state manually, but that seems very wrong.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)