[jira] [Created] (KAFKA-15834) Subscribing to non-existent topic blocks StreamThread from stopping

["Greg Harris (Jira)" <ji...@apache.org>]

Greg Harris created KAFKA-15834:
-----------------------------------

             Summary: Subscribing to non-existent topic blocks StreamThread from stopping
                 Key: KAFKA-15834
                 URL: https://issues.apache.org/jira/browse/KAFKA-15834
             Project: Kafka
          Issue Type: Bug
          Components: streams
    Affects Versions: 3.6.0
            Reporter: Greg Harris


In NamedTopologyIntegrationTest#shouldContinueProcessingOtherTopologiesWhenNewTopologyHasMissingInputTopics a topology is created which references an input topic which does not exist. The test as-written passes, but the KafkaStreams#close(Duration) at the end times out, and leaves StreamsThreads running.

From some cursory investigation it appears that this is happening:
1. The consumer calls the StreamsPartitionAssignor, which calls TaskManager#handleRebalanceStart as a side-effect
2. handleRebalanceStart sets the rebalanceInProgress flag
3. This flag is checked by StreamThread.runLoop, and causes the loop to remain running.
4. The consumer never calls StreamsRebalanceListener#onPartitionsAssigned, because the topic does not exist
5. Because no partitions are ever assigned, the TaskManager#handleRebalanceComplete never clears the rebalanceInProgress flag
 
This log message is printed in a tight loop while the close is ongoing and the consumer is being polled with zero duration:
{noformat}
[2023-11-15 11:42:43,661] WARN [Consumer clientId=NamedTopologyIntegrationTestshouldContinueProcessingOtherTopologiesWhenNewTopologyHasMissingInputTopics-942756f8-5213-4c44-bb6b-5f805884e026-StreamThread-1-consumer, groupId=NamedTopologyIntegrationTestshouldContinueProcessingOtherTopologiesWhenNewTopologyHasMissingInputTopics] Received unknown topic or partition error in fetch for partition unique_topic_prefix-topology-1-store-repartition-0 (org.apache.kafka.clients.consumer.internals.FetchCollector:321)
{noformat}
Practically, this means that this test leaks two StreamsThreads and the associated clients and sockets, and delays the completion of the test until the KafkaStreams#close(Duration) call times out.

Either we should change the rebalanceInProgress flag to avoid getting stuck in this rebalance state, or figure out a way to shut down a StreamsThread that is in an extended rebalance state during shutdown.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)