CODEWITHSUNDEEP
javaapache-kafkaapache-kafka-streams
Hessam
Hessam

Reputation: 1415

KafkaStreams application stops working with message Unsubscribed all topics or patterns and assigned partitions

I have a KafkaStream application that used to work perfectly. Now no matter how many times I restart it with new Application id it never start consuming the topic and I receive this log:

INFO org.apache.kafka.streams.KafkaStreams stream-client [score_redeemX-7e37d43e-12ab-4c66-984e-5f959d4e5e08] State transition from REBALANCING to RUNNING
INFO org.apache.kafka.clients.consumer.KafkaConsumer [Consumer clientId=score_redeemX-7e37d43e-12ab-4c66-984e-5f959d4e5e08-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions
INFO org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeemX-7e37d43e-12ab-4c66-984e-5f959d4e5e08-StreamThread-1] State transition from PARTITIONS_ASSIGNED to RUNNING
INFO org.apache.kafka.clients.consumer.KafkaConsumer [Consumer clientId=score_redeemX-7e37d43e-12ab-4c66-984e-5f959d4e5e08-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions

When I set log level to debug this is the log the application produces:

2020-05-09 18:45:20,335 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:20,338 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:20,339 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:21,339 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:21,341 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:21,342 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:22,343 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:22,344 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:22,345 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:22,435 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer, groupId=score_redeem] Sending Heartbeat request to coordinator ###.###.###.###:9092 (id: 2147483642 rack: null)
2020-05-09 18:45:22,437 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer, groupId=score_redeem] Received successful Heartbeat response
2020-05-09 18:45:22,465 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer, groupId=score_redeem] Sending Heartbeat request to coordinator ###.###.###.###:9092 (id: 2147483642 rack: null)
2020-05-09 18:45:22,466 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer, groupId=score_redeem] Received successful Heartbeat response
2020-05-09 18:45:22,468 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1-consumer, groupId=score_redeem] Sending Heartbeat request to coordinator ###.###.###.###:9092 (id: 2147483642 rack: null)
2020-05-09 18:45:22,470 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1-consumer, groupId=score_redeem] Received successful Heartbeat response

No task is assigned as the log shows:

2020-05-09 18:43:04,322 DEBUG org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer] Constructed client metadata {0633c837-8b94-4ba3-8e59-67a4c71de337=ClientMetadata{hostInfo=null, consumers=[score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer-b830a8e5-8654-40b7-8142-d1754c32268a], state=[activeTasks: ([]) standbyTasks: ([]) assignedTasks: ([]) prevActiveTasks: ([]) prevStandbyTasks: ([]) prevAssignedTasks: ([]) capacity: 1]}} from the member subscriptions.

This is the part of log that shows information about leader and member:

2020-05-09 18:43:04,361 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer, groupId=score_redeem] Received successful JoinGroup response: JoinGroupResponse(throttleTimeMs=0, error=NONE, generationId=4, groupProtocol=stream, memberId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer-93905915-8b52-44b4-afd1-2a5c79a272a4, leaderId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer-b830a8e5-8654-40b7-8142-d1754c32268a, members=)
2020-05-09 18:43:04,362 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer, groupId=score_redeem] Sending follower SyncGroup to coordinator ###.###.###.###:9092 (id: 2147483642 rack: null): (type=SyncGroupRequest, groupId=score_redeem, generationId=4, memberId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer-93905915-8b52-44b4-afd1-2a5c79a272a4, groupAssignment=)

I have many applications running and all of them are now experiencing this. I tried making new connectors and still no luck. But I can successfully consume from those topics with kafka-console-consumer.

Another strange yet very important point is when I rewrote the app with Kafka Consumer API it worked fine!!!

Upvotes: 3

Views: 6670

Answers (1)

Matthias J. Sax
Matthias J. Sax

Reputation: 62330

As mentioned by @bbejeck already, the StreamThread-1-restore-consumer logs Unsubscribed all topics or patterns and assigned partitions and it's an INFO level log. It just mark the end of a potential state restore and is not related to reading input topics.

The consumer heartbeat logs just document expected behavior.

The other logs: Committed all active tasks [] and standby tasks [] in 0ms might indicate that there are no active tasks. This would explain why there is no progress. However, it raised the question why are there no tasks.

During a rebalance the task assignment is logged per instance. Can you confirm that no tasks are assigned? Furthermore, one Kafka Streams instance (per application) will act as "group leader" and thus log even more details about the assignment it computes. (Btw: what version are you using?)

Upvotes: 3

Related Questions