With EOS enabled , Reinitializing stateStores get an NPE because checkpoint is null.
How to reproduce
- changelog topic with short `retention.ms` and `delete` policy (just to reproduce the symptom easily)
- exaclty once semantic enabled
- no cleanup
- two task[0_0],[0,1] , two Spring Boot (assign was#1:task[0_0], was#2:task[0_1])
- write some data each state store(changelog topic will soon erase those messages. by short "retentin.ms")
- when was#2 is killed, then was#1 will restore task[0_1]'s data on its own rocksDB
- In the process, it finds a checkpoint and an error occurs.(AbstractStateManager #66)
When EOS is enabled, the checkpoint will be null.
I think , need to add some code to create a Checkpoint.