[ https://issues.apache.org/jira/browse/KAFKA-6860?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16508317#comment-16508317 ]
Guozhang Wang commented on KAFKA-6860: -------------------------------------- I see. Thanks for the explanation [~mjsax]. And your proposed fix makes sense to me. I think a more general solution would involve also fixing the double checkpointing for non-EOS case: today we checkpoint in `suspend` if EOS is not turned in and in `closeSuspended` always. So for EOS, we only checkpoint in `closeSuspended`, while in non EOS we checkpoint in both, hence we have unnecessarily written twice of the checkpoints when closing. But for this general fix, I think it may be better to consider fixing with some refactoring on the ProcessorStateManager code, and hence not necessarily to be included for this JIRA. > NPE when reinitializeStateStores with eos enabled > ------------------------------------------------- > > Key: KAFKA-6860 > URL: https://issues.apache.org/jira/browse/KAFKA-6860 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 1.1.0 > Environment: mac, kafka1.1 > Reporter: ko byoung kwon > Priority: Major > Original Estimate: 2h > Remaining Estimate: 2h > > *Symptom* > With EOS enabled , Reinitializing stateStores get an NPE because checkpoint > is null. > {code:java} > 2018-05-02 18:05:17.156 ERROR 60836 --- [-StreamThread-1] > o.a.k.s.p.internals.StreamThread : stream-thread > [kafka-stream-application-d6ec1dfb-9b7f-42dd-8b28-899ff3d1ad98-StreamThread-1] > Encountered the following error during processing: > java.lang.NullPointerException: null > at > org.apache.kafka.streams.processor.internals.AbstractStateManager.reinitializeStateStoresForPartitions(AbstractStateManager.java:66) > ~[kafka-streams-1.1.0.jar:na] > at > org.apache.kafka.streams.processor.internals.ProcessorStateManager.reinitializeStateStoresForPartitions(ProcessorStateManager.java:155) > ~[kafka-streams-1.1.0.jar:na] > at > org.apache.kafka.streams.processor.internals.AbstractTask.reinitializeStateStoresForPartitions(AbstractTask.java:230) > ~[kafka-streams-1.1.0.jar:na] > at > org.apache.kafka.streams.processor.internals.StoreChangelogReader.restore(StoreChangelogReader.java:94) > ~[kafka-streams-1.1.0.jar:na] > {code} > *How to reproduce* > *configure as* > - changelog topic with short `retention.ms` and `delete` policy (just to > reproduce the symptom easily) > ex) retention.ms=30000,cleanup.policy=deleteĀ > - exaclty once semantic enabled > - no cleanup > *Step* > - 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) > {code:java} > // My code > Map<String, String> topicConfiguration = new HashMap<>(); > topicConfiguration.putIfAbsent("cleanup.policy", "delete"); > topicConfiguration.putIfAbsent("file.delete.delay.ms", "0"); > topicConfiguration.putIfAbsent("retention.ms", "3000"); > builder.stream(properties.getSourceTopic(), > Consumed.with(Serdes.Long(), Serdes.String())) > .groupByKey() > .count(Materialized > .<Long, Long, KeyValueStore<Bytes, > byte[]>>as(ORDER_STORE_NAME) > .withKeySerde(Serdes.Long()) > .withValueSerde(Serdes.Long()) > .withLoggingEnabled(topicConfiguration)); > {code} > *Suggestion* > When EOS is enabled, the checkpoint will be null. > I think , need to add some code to create a Checkpoint. > As follows > {code:java} > // # At org.apache.kafka.streams.processor.internals.AbstractStateManager #66 > // # suggestion start > if (checkpoint == null) { > checkpoint = new OffsetCheckpoint(new File(baseDir, > CHECKPOINT_FILE_NAME)); > } > // # suggestion end > try { > checkpoint.write(checkpointableOffsets); > } catch (final IOException fatalException) { > log.error("Failed to write offset checkpoint file to {} while > re-initializing {}: {}", checkpoint, stateStores, fatalException); > throw new StreamsException("Failed to reinitialize global store.", > fatalException); > } > {code} -- This message was sent by Atlassian JIRA (v7.6.3#76005)