[KAFKA-4539] StreamThread is not correctly creating StandbyTasks

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.10.2.0
• Fix Version/s: 0.10.2.0
• Component/s: streams
• Assignee: Damian Guy
• Reporter: Damian Guy
• Created: 14/Dec/16 12:26
• Updated: 16/Dec/16 01:23
• Resolved: 16/Dec/16 01:23

Description

Fails because createStandbyTask(..) can return null if the topology for the TaskId doesn't have any state stores.

[2016-12-14 12:20:29,758] ERROR [StreamThread-1] User provided listener org.apache.kafka.streams.processor.internals.StreamThread$1 for group kafka-music-charts failed on partition assignment (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
java.lang.NullPointerException
	at org.apache.kafka.streams.processor.internals.StreamThread$StandbyTaskCreator.createTask(StreamThread.java:1241)
	at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:1188)
	at org.apache.kafka.streams.processor.internals.StreamThread.addStandbyTasks(StreamThread.java:915)
	at org.apache.kafka.streams.processor.internals.StreamThread.access$400(StreamThread.java:72)
	at org.apache.kafka.streams.processor.internals.StreamThread$1.onPartitionsAssigned(StreamThread.java:239)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:230)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:314)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:278)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:261)
	at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1022)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:987)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:568)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:357)

Also fails because the checkpointedOffsets from the newly created StandbyTask aren't added to the offsets map, so the partitions don't get assigned. We then get:

Issue Links

• links to: GitHub Pull Request #2255