You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "eduwercamacaro (via GitHub)" <gi...@apache.org> on 2025/10/22 16:06:31 UTC
[PR] REFACTOR: Startup state stores initialization [kafka]
eduwercamacaro opened a new pull request, #20749:
URL: https://github.com/apache/kafka/pull/20749
This pull request suggests adding a new stage for the state store lifecycle.
Since we don't open and/or initialize temporary tasks, we can fix the issue described in this issue: KAFKA-19434 by modifying the state store refactor. It is now possible to pre-initialize state stores and make them readable, which also is required by KIP-1035 implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] REFACTOR: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2455830356
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -221,44 +228,23 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
.map(t -> new TopicPartition(t, id.partition()))
.collect(Collectors.toSet());
final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
- id,
- eosEnabled,
- logContext,
- this,
- subTopology.storeToChangelogTopic(),
- inputPartitions,
- stateUpdaterEnabled
+ id,
+ eosEnabled,
+ logContext,
+ this,
+ subTopology.storeToChangelogTopic(),
+ inputPartitions,
+ stateUpdaterEnabled
);
-
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
- try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ final StartupContext initContext = new StartupContext(id, config, stateManager);
+ // TODO: we need to pass a proper logPrefix
+ StateManagerUtil.registerStateStores(log, "", subTopology, stateManager, this, initContext);
+ for (final StateStore stateStore : subTopology.stateStores()) {
+ if (!stateStore.isOpen()) {
Review Comment:
Yes, this needs to be updated for the other stores.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2698943190
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -221,44 +228,23 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
.map(t -> new TopicPartition(t, id.partition()))
.collect(Collectors.toSet());
final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
- id,
- eosEnabled,
- logContext,
- this,
- subTopology.storeToChangelogTopic(),
- inputPartitions,
- stateUpdaterEnabled
+ id,
+ eosEnabled,
+ logContext,
+ this,
+ subTopology.storeToChangelogTopic(),
+ inputPartitions,
+ stateUpdaterEnabled
);
-
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
- try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ final StartupContext initContext = new StartupContext(id, config, stateManager);
+ // TODO: we need to pass a proper logPrefix
+ StateManagerUtil.registerStateStores(log, "", subTopology, stateManager, this, initContext);
+ for (final StateStore stateStore : subTopology.stateStores()) {
+ if (!stateStore.isOpen()) {
Review Comment:
Thanks @Nikita-Shupletsov
There are many state stores that needed to be updated with the preInit implementation. I pushed [this commit](https://github.com/apache/kafka/pull/20749/commits/b4cb0ff7bba4218251cdf4490edca0b2afd665d0) to address this comment.
Let me know what you about it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on PR #20749:
URL: https://github.com/apache/kafka/pull/20749#issuecomment-3847605682
@bbejeck I modified this PR according to our last discussion about the solution for [KAFKA-19434](https://issues.apache.org/jira/browse/KAFKA-19434)
We basically open the stores found in the state directory during initialization and then close them.
cc: @mjsax
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #20749:
URL: https://github.com/apache/kafka/pull/20749#issuecomment-3935786236
> looks like build is failing after this commit
Thanks @omkreddy! I've pushed a fix we should be good now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on PR #20749:
URL: https://github.com/apache/kafka/pull/20749#issuecomment-3935665407
looks like build is failing after this commit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2770932903
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -237,35 +252,18 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
inputPartitions
);
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
+ final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
+ } catch (final TaskCorruptedException tce) {
+ log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
Review Comment:
`registerStateStores` also can throw a `StreamsException`, `LockException` , or a `ProcessorStateException` should we handle these exceptions as well? If not add some comments on the reasoning
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -237,35 +252,18 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
inputPartitions
);
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
+ final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
+ } catch (final TaskCorruptedException tce) {
+ log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
+ } finally {
+ // Make sure the state manager writes the local checkpoint file before closing the stores
+ // This will be replaced in the future when removing the checkpoint file dependency.
+ temporaryStateManager.checkpoint();
Review Comment:
Same as above `temporaryStateManager.checkpoint()` can throw a `ProcessorStateException` but we don't handle it here, meaning `close()` is not called. So do we consider this a fatal exception and allow it to bubble up and kill the main thread? If not, I'm guessing we'll need to catch this exception and do something. Either way let's add a couple of comments on the handling
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -237,35 +252,18 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
inputPartitions
);
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
+ final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
+ } catch (final TaskCorruptedException tce) {
+ log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
+ } finally {
+ // Make sure the state manager writes the local checkpoint file before closing the stores
+ // This will be replaced in the future when removing the checkpoint file dependency.
+ temporaryStateManager.checkpoint();
+ temporaryStateManager.close();
Review Comment:
I think we also need to add `StateDirectory.unlock` here too.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -237,35 +252,18 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
inputPartitions
);
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
+ final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
+ } catch (final TaskCorruptedException tce) {
+ log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
+ } finally {
+ // Make sure the state manager writes the local checkpoint file before closing the stores
+ // This will be replaced in the future when removing the checkpoint file dependency.
+ temporaryStateManager.checkpoint();
+ temporaryStateManager.close();
}
+ tasksForLocalState.put(id, new StartupState(id, subTopology, temporaryStateManager));
Review Comment:
`temporaryStateManager` is closed above line 264 but it's passed here to `StartupState`.
It's never accessed from `StartupState` . With the new work is `StartupState` required or can we get away with `Set<TaskId>` ? Or is it going to be needed for subsequent work?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] REFACTOR: Startup state stores initialization [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2453535667
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -221,44 +228,23 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
.map(t -> new TopicPartition(t, id.partition()))
.collect(Collectors.toSet());
final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
- id,
- eosEnabled,
- logContext,
- this,
- subTopology.storeToChangelogTopic(),
- inputPartitions,
- stateUpdaterEnabled
+ id,
+ eosEnabled,
+ logContext,
+ this,
+ subTopology.storeToChangelogTopic(),
+ inputPartitions,
+ stateUpdaterEnabled
);
-
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
- try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ final StartupContext initContext = new StartupContext(id, config, stateManager);
+ // TODO: we need to pass a proper logPrefix
+ StateManagerUtil.registerStateStores(log, "", subTopology, stateManager, this, initContext);
+ for (final StateStore stateStore : subTopology.stateStores()) {
+ if (!stateStore.isOpen()) {
Review Comment:
I assume, the rest of stores will be updated later? right now there is only one Store updated: RockDBStore. I opened a random store(InMemoryKeyValueStore) and it sets the open field to true in the init method, so it will start failing here, I think
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2682836819
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -230,44 +238,23 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
.map(t -> new TopicPartition(t, id.partition()))
.collect(Collectors.toSet());
final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
- id,
- eosEnabled,
- logContext,
- this,
- subTopology.storeToChangelogTopic(),
- inputPartitions,
- stateUpdaterEnabled
+ id,
+ eosEnabled,
+ logContext,
+ this,
+ subTopology.storeToChangelogTopic(),
+ inputPartitions,
+ stateUpdaterEnabled
);
-
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
- try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ final StartupContext initContext = new StartupContext(id, config, stateManager);
+ // TODO: we need to pass a proper logPrefix
Review Comment:
I think we need to use a different format for the logPrefix since this method is not being invoked by a StreamThread. Just pushed [this commit ](https://github.com/apache/kafka/pull/20749/commits/f8febe71461adae7288943382a27ba9f91436a46)let me know what you think about it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2611679375
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -230,44 +238,23 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
.map(t -> new TopicPartition(t, id.partition()))
.collect(Collectors.toSet());
final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
- id,
- eosEnabled,
- logContext,
- this,
- subTopology.storeToChangelogTopic(),
- inputPartitions,
- stateUpdaterEnabled
+ id,
+ eosEnabled,
+ logContext,
+ this,
+ subTopology.storeToChangelogTopic(),
+ inputPartitions,
+ stateUpdaterEnabled
);
-
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
- try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ final StartupContext initContext = new StartupContext(id, config, stateManager);
+ // TODO: we need to pass a proper logPrefix
Review Comment:
instead of `""` we could use `logPrefix()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2699551327
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -221,44 +228,23 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
.map(t -> new TopicPartition(t, id.partition()))
.collect(Collectors.toSet());
final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
- id,
- eosEnabled,
- logContext,
- this,
- subTopology.storeToChangelogTopic(),
- inputPartitions,
- stateUpdaterEnabled
+ id,
+ eosEnabled,
+ logContext,
+ this,
+ subTopology.storeToChangelogTopic(),
+ inputPartitions,
+ stateUpdaterEnabled
);
-
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
- try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ final StartupContext initContext = new StartupContext(id, config, stateManager);
+ // TODO: we need to pass a proper logPrefix
+ StateManagerUtil.registerStateStores(log, "", subTopology, stateManager, this, initContext);
+ for (final StateStore stateStore : subTopology.stateStores()) {
+ if (!stateStore.isOpen()) {
Review Comment:
@eduwercamacaro
thanks!
just to confirm: the idea is that we want the owner of a store to be responsible for its proper initialization, right? so if KS hasn't called preInit, they will need to do that themselves during init
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #20749:
URL: https://github.com/apache/kafka/pull/20749#issuecomment-3729980635
@eduwercamacaro - you can ignore my comments on testing - we can do as a follow-up PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on PR #20749:
URL: https://github.com/apache/kafka/pull/20749#issuecomment-3661064968
@bbejeck Thanks for your time reviewing this. I already rebased this branch.
I believe [this integration test](https://github.com/apache/kafka/pull/20749/files#diff-a8f919d43ec1cebdb136acf94adfbeb20fef0cda0ad5d29b7010e0186b355d71R287) is already covering that case. what do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2699090678
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -333,25 +333,19 @@ private void closeDirtyAndRevive(final Collection<Task> taskWithChangelogs, fina
}
}
- private Map<Task, Set<TopicPartition>> assignStartupTasks(final Map<TaskId, Set<TopicPartition>> tasksToAssign,
- final String threadLogPrefix,
- final TopologyMetadata topologyMetadata,
Review Comment:
This PR only removes the `threadLodPrefix` parameter. It now uses the `logPrefix` instance variable. [see this commit](https://github.com/apache/kafka/pull/20749/commits/49e786f4f7379c57366b1660454f148955fed941)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #20749:
URL: https://github.com/apache/kafka/pull/20749#issuecomment-3935267444
Merged #20749 into trunk
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] REFACTOR: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2455820031
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java:
##########
@@ -157,9 +159,14 @@ public RocksDBStore(final String name,
@Override
public void init(final StateStoreContext stateStoreContext,
final StateStore root) {
+ initialized.set(true);
// open the DB dir
metricsRecorder.init(metricsImpl(stateStoreContext), stateStoreContext.taskId());
- openDB(stateStoreContext.appConfigs(), stateStoreContext.stateDir());
+ if (!open) {
+ preInit(stateStoreContext);
Review Comment:
Thank you for your time reviewing this PR.
That is a valid concern.
As things are right now in this PR, it's not clear when each method in the state store lifecycle is called. This makes things a lot harder for the people who write state stores, including the existing `RocksDBStore`.
I agree that we need to make the implementation a little harder on our end and a little easier on the specific state stores’ end.
I was thinking that we could document the `preInit` method and specify that it will be called immediately after the state store is built (in the topology builder), allowing state stores to implement logic for opening resources that will be required during the `init` phase. In the case of the `RocksDBStore`, we will open the store during this phase.
Although I am not familiar with custom state stores, I would assume that they could also open a database connection during this phase.
Since the state store was already opened during the `preInit` phase, the `init` method can only concentrate on preparing it for use in StreamThread processing (i.e., registering metrics in the process context). This is because the init method will run during the rebalance process.
State store authors would find it easy to implement or evolve their custom implementations with a more clear lifecycle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2623694213
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -230,44 +238,23 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
.map(t -> new TopicPartition(t, id.partition()))
.collect(Collectors.toSet());
final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
- id,
- eosEnabled,
- logContext,
- this,
- subTopology.storeToChangelogTopic(),
- inputPartitions,
- stateUpdaterEnabled
+ id,
+ eosEnabled,
+ logContext,
+ this,
+ subTopology.storeToChangelogTopic(),
+ inputPartitions,
+ stateUpdaterEnabled
);
-
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
- try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ final StartupContext initContext = new StartupContext(id, config, stateManager);
+ // TODO: we need to pass a proper logPrefix
Review Comment:
I'm not totally sure if that's what we want because the `logPrefix()` method will use a `stream-thread-` as prefix. I think we will need to provide a log prefix when assigning these startup state store to a Task.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2621100614
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -205,7 +212,8 @@ public ProcessorStateManager(final TaskId taskId,
final ChangelogRegister changelogReader,
final Map<String, String> storeToChangelogTopic,
final Collection<TopicPartition> sourcePartitions,
- final boolean stateUpdaterEnabled) throws ProcessorStateException {
+ final boolean stateUpdaterEnabled,
+ final boolean startupState) throws ProcessorStateException {
Review Comment:
Mostly, this boolean flag controls when to call the `preInit` and the `init` methods within this class. This is because the `ProcessorStateManager` is the one that controls the state store lifecycle.
But I think it is probably better to create a new method that specifically registers startup states. I made some changes that removes this field by adding a new method in the ProcessorStateManager. see [this commit](https://github.com/apache/kafka/pull/20749/commits/fd93623864f3ae78edab8ac876ded56a495b21ac)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck merged PR #20749:
URL: https://github.com/apache/kafka/pull/20749
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2682049133
##########
streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java:
##########
@@ -71,6 +71,8 @@ public interface StateStore {
*/
void init(final StateStoreContext stateStoreContext, final StateStore root);
+ default void preInit(final StateStoreContext stateStoreContext) {}
+
Review Comment:
This is the part of KIP-1035 that needs to be changed so that this new method can be added to the StateStore interface.
`preInit` essentially adds a new callback for StateStores that allows us to initialize resources before the StateStore is assigned to a `StreamThread` via `StateStore#init`. For the case of RocksDBStateStore, we will load current offsets from the ColumnFamily.
On this PR, we need this new method because it allows the StateDirectory class to open the state stores without assigning them to a StreamThread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #20749:
URL: https://github.com/apache/kafka/pull/20749#issuecomment-3745628120
@eduwercamacaro the failures are related - I think [addressing this comment](https://github.com/apache/kafka/pull/20749#discussion_r2453535667) should fix the integration tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2674152338
##########
streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java:
##########
@@ -71,6 +71,8 @@ public interface StateStore {
*/
void init(final StateStoreContext stateStoreContext, final StateStore root);
+ default void preInit(final StateStoreContext stateStoreContext) {}
+
Review Comment:
Missed this before - what's the functionality of `preInit` vs. `init`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] REFACTOR: Startup state stores initialization [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2453535667
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -221,44 +228,23 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
.map(t -> new TopicPartition(t, id.partition()))
.collect(Collectors.toSet());
final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
- id,
- eosEnabled,
- logContext,
- this,
- subTopology.storeToChangelogTopic(),
- inputPartitions,
- stateUpdaterEnabled
+ id,
+ eosEnabled,
+ logContext,
+ this,
+ subTopology.storeToChangelogTopic(),
+ inputPartitions,
+ stateUpdaterEnabled
);
-
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
- try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ final StartupContext initContext = new StartupContext(id, config, stateManager);
+ // TODO: we need to pass a proper logPrefix
+ StateManagerUtil.registerStateStores(log, "", subTopology, stateManager, this, initContext);
+ for (final StateStore stateStore : subTopology.stateStores()) {
+ if (!stateStore.isOpen()) {
Review Comment:
I assume, the rest of stores will be updated later? right now there is only one Store is updated: RockDBStore. I opened a random store(InMemoryKeyValueStore) and it sets the open field to true in the init method, so it will start failing here, I think
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] REFACTOR: Startup state stores initialization [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2453532998
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java:
##########
@@ -157,9 +159,14 @@ public RocksDBStore(final String name,
@Override
public void init(final StateStoreContext stateStoreContext,
final StateStore root) {
+ initialized.set(true);
// open the DB dir
metricsRecorder.init(metricsImpl(stateStoreContext), stateStoreContext.taskId());
- openDB(stateStoreContext.appConfigs(), stateStoreContext.stateDir());
+ if (!open) {
+ preInit(stateStoreContext);
Review Comment:
maybe it's not a valid concern, but it looks like this interface would be hard to explain and thus use correctly: there is a two stage init process. sometimes we call preInit(phase one) explicitly, but there are a lot of cases when the author of the store needs to call it themselves in the init(phase two).
as we rely on stateStore.isOpen be true, maybe we should communicate that and follow the same path every time we want to initialize a store?
i.e.
* phase one(preInit/open)
* check if opened
* phase two(init)
it will slightly complicate the implementation on our side, but will make it significantly easier to follow for users
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "eduwercamacaro (via GitHub)" <gi...@apache.org>.
eduwercamacaro commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2788115921
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -237,35 +252,18 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
inputPartitions
);
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
+ final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
+ } catch (final TaskCorruptedException tce) {
+ log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
+ } finally {
+ // Make sure the state manager writes the local checkpoint file before closing the stores
+ // This will be replaced in the future when removing the checkpoint file dependency.
+ temporaryStateManager.checkpoint();
Review Comment:
I moved this logic inside the try/catch block to ensure that we always call the `close()` method. Any exception from the `close()` method is considered fatal and will break the main thread.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -237,35 +252,18 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
inputPartitions
);
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
+ final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
+ } catch (final TaskCorruptedException tce) {
+ log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
+ } finally {
+ // Make sure the state manager writes the local checkpoint file before closing the stores
+ // This will be replaced in the future when removing the checkpoint file dependency.
+ temporaryStateManager.checkpoint();
+ temporaryStateManager.close();
Review Comment:
Currently, it is not needed because we keep the lock on the main thread until it is assigned to a SthreadThread. TaskManager changes the ownership of the Task using 'StateDirectory.removeStartupState'.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -237,35 +252,18 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
inputPartitions
);
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
+ final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
+ } catch (final TaskCorruptedException tce) {
+ log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
+ } finally {
+ // Make sure the state manager writes the local checkpoint file before closing the stores
+ // This will be replaced in the future when removing the checkpoint file dependency.
+ temporaryStateManager.checkpoint();
+ temporaryStateManager.close();
}
+ tasksForLocalState.put(id, new StartupState(id, subTopology, temporaryStateManager));
Review Comment:
Yep. Excellent observation.
You are right, we can use `Set<TaskId>`` instead.
I completely removed the `StartupState` class and the process to close the stores because we are immediately closing them.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -237,35 +252,18 @@ public void initializeStartupTasks(final TopologyMetadata topologyMetadata,
inputPartitions
);
- final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
- id,
- config,
- stateManager,
- streamsMetrics,
- dummyCache
- );
-
- final Task task = new StandbyTask(
- id,
- inputPartitions,
- subTopology,
- topologyMetadata.taskConfig(id),
- streamsMetrics,
- stateManager,
- this,
- dummyCache,
- context
- );
-
+ final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
try {
- task.initializeIfNeeded();
-
- tasksForLocalState.put(id, task);
- } catch (final TaskCorruptedException e) {
- // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it
- task.suspend();
- task.closeDirty();
+ StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
+ } catch (final TaskCorruptedException tce) {
+ log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
Review Comment:
Good point.
I added some comments and JavaDoc for this method.
Those exceptions are considered fatal and will break the StreamThread. We only handle `TaskCorruptedException` by logging and continuing with the initialization as this exception will be handled by the StreamThread later in the first assignment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19434: Startup state stores initialization [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #20749:
URL: https://github.com/apache/kafka/pull/20749#discussion_r2604522491
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -255,6 +279,10 @@ void assignToStreamThread(final LogContext logContext,
this.sourcePartitions.addAll(sourcePartitions);
}
+ void reuseState() {
Review Comment:
What are the semantics around `reuseState`? Seems a little odd that calling `reuse` sets false, probably could use a small comment or a method rename.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -205,7 +212,8 @@ public ProcessorStateManager(final TaskId taskId,
final ChangelogRegister changelogReader,
final Map<String, String> storeToChangelogTopic,
final Collection<TopicPartition> sourcePartitions,
- final boolean stateUpdaterEnabled) throws ProcessorStateException {
+ final boolean stateUpdaterEnabled,
+ final boolean startupState) throws ProcessorStateException {
Review Comment:
What does `startupState` represent, maybe rename to something like `hasStateOnStartup`?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -333,25 +333,19 @@ private void closeDirtyAndRevive(final Collection<Task> taskWithChangelogs, fina
}
}
- private Map<Task, Set<TopicPartition>> assignStartupTasks(final Map<TaskId, Set<TopicPartition>> tasksToAssign,
- final String threadLogPrefix,
- final TopologyMetadata topologyMetadata,
Review Comment:
Why did these parameters go away?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -220,6 +228,22 @@ public ProcessorStateManager(final TaskId taskId,
this.checkpointFile = new OffsetCheckpoint(stateDirectory.checkpointFileFor(taskId));
log.debug("Created state store manager for task {}", taskId);
+ this.startupState = new AtomicBoolean(startupState);
Review Comment:
same would apply to the `AtomicBoolean` variable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org