Kafka Stream: Cannot invoke "org.apache.kafka.streams.processor.internals.SourceNode.getTimestampExtractor()" because "source" is null

[An, Hongguo (CORP)]

Hi:
Every time, I restart my kafka stream app, it failed, and I have to reset the 
app, the error is:

org.apache.kafka.streams.errors.StreamsException: stream-thread 
[microapi-unified-profile-data-sync.dit-1067669c-2364-440d-a1c3-69ad45cc301d-StreamThread-4]
 Failed to rebalance.
                at 
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:972)
                at 
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:863)
                at 
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:819)
                at 
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:788)
Caused by: java.lang.NullPointerException: Cannot invoke 
"org.apache.kafka.streams.processor.internals.SourceNode.getTimestampExtractor()"
 because "source" is null
                at 
org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:230)
                at 
org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:172)
                at 
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:459)
                at 
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:410)
                at 
org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:395)
                at 
org.apache.kafka.streams.processor.internals.TaskManager.addStreamTasks(TaskManager.java:148)
                at 
org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:107)
                at 
org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:294)
                at 
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:285)
                at 
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:424)
                at 
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:358)
                at 
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:353)
                at 
org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1251)
                at 
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1220)
                at 
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1159)
                at 
brave.kafka.clients.TracingConsumer.poll(TracingConsumer.java:84)
                at 
brave.kafka.clients.TracingConsumer.poll(TracingConsumer.java:78)
                at 
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:963)

It works after reset, please help.

Thanks
Andrew


This message and any attachments are intended only for the use of the addressee 
and may contain information that is privileged and confidential. If the reader 
of the message is not the intended recipient or an authorized representative of 
the intended recipient, you are hereby notified that any dissemination of this 
communication is strictly prohibited. If you have received this communication 
in error, notify the sender immediately by return email and delete the message 
and any attachments from your system.