[jira] [Created] (KAFKA-7213) NullPointerException during state restoration in kafka streams

["Abhishek Agarwal (JIRA)" <ji...@apache.org>]

Abhishek Agarwal created KAFKA-7213:
---------------------------------------

             Summary: NullPointerException during state restoration in kafka streams
                 Key: KAFKA-7213
                 URL: https://issues.apache.org/jira/browse/KAFKA-7213
             Project: Kafka
          Issue Type: Bug
          Components: streams
    Affects Versions: 1.0.0
            Reporter: Abhishek Agarwal
            Assignee: Abhishek Agarwal


I had written a custom state store which has a batch restoration callback registered. What I have observed, when multiple consumer instances are restarted, the application keeps failing with NullPointerException. The stack trace is 
{noformat}
java.lang.NullPointerException: null
	at org.apache.kafka.streams.state.internals.RocksDBStore.putAll(RocksDBStore.java:351) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.state.internals.RocksDBSlotKeyValueBytesStore.putAll(RocksDBSlotKeyValueBytesStore.java:100) ~[streams-core-1.0.0.297.jar:?]
	at org.apache.kafka.streams.state.internals.RocksDBSlotKeyValueBytesStore$SlotKeyValueBatchRestoreCallback.restoreAll(RocksDBSlotKeyValueBytesStore.java:303) ~[streams-core-1.0.0.297.jar:?]
	at org.apache.kafka.streams.processor.internals.CompositeRestoreListener.restoreAll(CompositeRestoreListener.java:89) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.processor.internals.StateRestorer.restore(StateRestorer.java:75) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.processor.internals.StoreChangelogReader.processNext(StoreChangelogReader.java:277) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.processor.internals.StoreChangelogReader.restorePartition(StoreChangelogReader.java:238) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.processor.internals.StoreChangelogReader.restore(StoreChangelogReader.java:83) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.processor.internals.TaskManager.updateNewAndRestoringTasks(TaskManager.java:263) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:803) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774) ~[kafka-streams-1.0.0.jar:?]
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744) ~[kafka-streams-1.0.0.jar:?]
{noformat}

The faulty line in question is 
{noformat}
db.write(wOptions, batch);
{noformat}

in RocksDBStore.java which would mean that db variable is null. Probably the store has been closed and restoration is still being done on it. After going through the code, I think the problem is when state transitions from PARTITIONS_ASSIGNED to PARTITIONS_REVOKED and restoration is still in progress. 
In such state transition, while the active tasks themselves are closed, the changelog reader is not reset. It tries to restore the tasks that have already been closed, db is null and results in NPE. 

I will put in a fix to see if that fixes the issue. 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)