NPE at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:198)

[Levani Kokhreidze]

Hi,

I’ve cleaned up some old topics of old kafka streams instances from kafka broker and as a result when I did restart of the services this exceptions started to happen during rebalance phase:

2018-11-14 10:33:03,825 ERROR [StreamThread][common-streaming-pipeline-v5-client-id-StreamThread-1][streaming-server] - stream-thread [common-streaming-pipeline-v5-client-id-StreamThread-1] Error caught during partition assignment, will abort the current process and re-throw at the end of rebalance: {}

 java.lang.NullPointerException: null#012 at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:198)#012 at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:146)#012 at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:436)#012 at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:388)#012 at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:373)#012 at org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:148)#012at org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:107)#012 at org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:270)#012 at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:283)#012 at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:422)#012 at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:352)#012 at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:337)#012 at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:333)#012 at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1218)#012 at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1175)#012at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1154)#012 at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:861)#012 at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:814)#012 at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)#012 at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)

weird thing is if I restart kafka streams multiple times application starts successfully. Any ideas?