KAFKA-19434: Startup state stores initialization by eduwercamacaro · Pull Request #20749 · apache/kafka

eduwercamacaro · 2025-10-22T16:06:28Z

Instead of creating Standby tasks from the state directory, we open the
local stores that exists in the state directory.

This resolves the issue raised in #KAFKA-19434
, where
store-specific metrics were being duplicated due to tasks being created
in the main thread and then assigned to a StreamThread.

Additionally, since we can now read the offsets from the store during
instance initialization, this clears the way for the implementation of
KIP-1035. As of now, the stores are loading the offsets from the
checkpoint file, but in a later PR, we will read these offsets from the
state store itself.

This PR modifies the behavior of Kafka Streams when initializing. Now
for each pre-existing store on the state directory: We open the store,
read offsets from the checkpoint file and then close it again. The
reason why we open the store is because the store will be responsible
for tracking the offsets and we will deprecate the checkpoint file.

Reviewers: Nikiita Shuplestovnshupletsov@confluent.io, Bill
Bejeckbbejeck@apache.org

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

Nikita-Shupletsov

Thanks a lot for putting it together!
A much better idea than what I had in my PR

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

bbejeck

Thanks @eduwercamacaro! I made an initial pass

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java

bbejeck · 2025-12-09T21:55:12Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java

        this.checkpointFile = new OffsetCheckpoint(stateDirectory.checkpointFileFor(taskId));

        log.debug("Created state store manager for task {}", taskId);
+        this.startupState = new AtomicBoolean(startupState);


same would apply to the AtomicBoolean variable

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

bbejeck

@eduwercamacaro I've made another pass and confirmed the fix by running a streams application and stopping/starting several times with existing state. Overall this lgtm
Can you rebase this PR?

Also I think we could use a new test in KafkaStreamsTelemetryIntegrationTest that start/stops/restarts a single instance and confirms the state metrics getting emitted.

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

eduwercamacaro · 2025-12-16T15:15:18Z

@bbejeck Thanks for your time reviewing this. I already rebased this branch.

I believe this integration test is already covering that case. what do you think?

bbejeck

Thanks again for the PR @eduwercamacaro!
I have one additional question I missed from the last review.

Also can you remove the ProcessorStateManager.assignToStreamThread method? it's not used anymore.

The ProcessorStateManagerTest could use better coverage of registerStartupStateStores and initializeStoreOffsetsFromCheckpoint

streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java

bbejeck · 2026-01-09T17:55:34Z

@eduwercamacaro - you can ignore my comments on testing - we can do as a follow-up PR

bbejeck · 2026-01-13T17:49:27Z

@eduwercamacaro the failures are related - I think addressing this comment should fix the integration tests.

…ing startup store pre-initialization

This reverts commit b4cb0ff.

…ialization

eduwercamacaro · 2026-02-04T13:55:23Z

@bbejeck I modified this PR according to our last discussion about the solution for KAFKA-19434

We basically open the stores found in the state directory during initialization and then close them.

cc: @mjsax

bbejeck

Thanks for the PR, @eduwercamacaro! Overall, it looks good with few comments.

Also, ProcessorStateManager.assignToStreamThread on line 237 is unused; can we get rid of it?

bbejeck · 2026-02-05T19:25:18Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

+                    } 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();


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

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.

bbejeck · 2026-02-05T20:05:15Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

-                        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());


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

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.

bbejeck · 2026-02-05T20:29:29Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

+                        // 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();


I think we also need to add StateDirectory.unlock here too.

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'.

bbejeck · 2026-02-05T20:48:25Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

+                        temporaryStateManager.checkpoint();
+                        temporaryStateManager.close();
                    }
+                    tasksForLocalState.put(id, new StartupState(id, subTopology, temporaryStateManager));


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?

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.

- track startup task ids via a simple Set and unlock them on close - remove startup-task handoff/close APIs and related tests - drop unused standby-thread assignment helper in

eduwercamacaro

@bbejeck Thanks for your comments
I pushed a new commit with the suggestions.

eduwercamacaro · 2026-02-10T13:44:57Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

+                        // 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();


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'.

eduwercamacaro · 2026-02-10T13:49:02Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

-                        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());


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.

eduwercamacaro · 2026-02-10T13:51:52Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

+                        temporaryStateManager.checkpoint();
+                        temporaryStateManager.close();
                    }
+                    tasksForLocalState.put(id, new StartupState(id, subTopology, temporaryStateManager));


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.

eduwercamacaro · 2026-02-10T13:57:15Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java

+                    } 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();


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.

- Remove startup state on statedir auto cleanup

bbejeck

Thanks @eduwercamacaro! LGTM just waiting for the build to complete

bbejeck · 2026-02-20T14:36:40Z

Merged #20749 into trunk

omkreddy · 2026-02-20T15:43:12Z

looks like build is failing after this commit

bbejeck · 2026-02-20T16:08:55Z

looks like build is failing after this commit

Thanks @omkreddy! I've pushed a fix we should be good now

refactor startup state stores initialization

23ce8e8

github-actions bot added triage PRs from the community streams labels Oct 22, 2025

eduwercamacaro mentioned this pull request Oct 22, 2025

Experiment kip 1035 #20709

Closed

Nikita-Shupletsov reviewed Oct 22, 2025

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java Outdated Show resolved Hide resolved

Nikita-Shupletsov reviewed Oct 22, 2025

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java Outdated Show resolved Hide resolved

github-actions bot removed the triage PRs from the community label Oct 23, 2025

eduwercamacaro added 2 commits October 27, 2025 11:49

merge trunk

2bfc943

Remove rocksdb temp files

8dc87f2

eduwercamacaro changed the title ~~REFACTOR: Startup state stores initialization~~ KAFKA-19434: Startup state stores initialization Nov 21, 2025

eduwercamacaro marked this pull request as ready for review November 21, 2025 16:20

eduwercamacaro added 2 commits November 21, 2025 12:01

Merge branch 'trunk' into refactor-startup-store-initialization

2b1e77b

Fix merge conflicts

0e9e40b

Nikita-Shupletsov mentioned this pull request Nov 21, 2025

KAFKA-19434: introduced assignThread method to be able to initialize thread specific resources. #20570

Closed

eduwercamacaro added 4 commits November 21, 2025 13:14

Replace RocksDBStore#openDB to StateStore#preInit

13c9e91

Remove unused classes

050e725

Discover startup stores during topology initialization

a81eaed

Allow closing state stores during preInit phase

4ad6553

bbejeck reviewed Dec 9, 2025

View reviewed changes

bbejeck reviewed Dec 11, 2025

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java Outdated Show resolved Hide resolved

eduwercamacaro added 2 commits December 15, 2025 17:39

Add a new method that register startup stores in a ProcessorStateManager

fd93623

Merge branch 'trunk' into refactor-startup-store-initialization

233dccf

bbejeck reviewed Jan 8, 2026

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java Outdated Show resolved Hide resolved

eduwercamacaro added 2 commits January 12, 2026 08:24

Merge branch 'trunk' into refactor-startup-store-initialization

abac041

Set a log prefix for startup stores during initialization

f8febe7

eduwercamacaro added 2 commits January 16, 2026 10:01

Properly assign a StreamThread to a ProcessorStateManager created dur…

49e786f

…ing startup store pre-initialization

Implement preInit method for InMemory stores

b4cb0ff

eduwercamacaro added 2 commits January 23, 2026 15:17

Revert "Implement preInit method for InMemory stores"

7b614ea

This reverts commit b4cb0ff.

wip

cd0dc0a

eduwercamacaro force-pushed the refactor-startup-store-initialization branch from 3cacb12 to 4451626 Compare February 2, 2026 15:47

Refactor: Open and close state stores during pre-existing stores init…

6078803

…ialization

eduwercamacaro force-pushed the refactor-startup-store-initialization branch from 4451626 to 6078803 Compare February 2, 2026 16:11

eduwercamacaro added 5 commits February 3, 2026 09:01

Refacttor: remove startup stores from state manager

97dea13

Revert changes in StandbyTaskCreator

cc8d4f1

rename methods

60d8aa9

Fix unit tests

65ffd3f

Remove unused method

525b280

Merge branch 'trunk' into refactor-startup-store-initialization

191e9de

bbejeck reviewed Feb 5, 2026

View reviewed changes

streams: simplify startup state handling and locking

2c0bb94

- track startup task ids via a simple Set and unlock them on close - remove startup-task handoff/close APIs and related tests - drop unused standby-thread assignment helper in

eduwercamacaro commented Feb 10, 2026

View reviewed changes

eduwercamacaro added 2 commits February 10, 2026 09:08

Merge branch 'trunk' into refactor-startup-store-initialization

a8efdea

- Rename unit tests related to startup states

a395b3d

- Remove startup state on statedir auto cleanup

bbejeck approved these changes Feb 19, 2026

View reviewed changes

bbejeck merged commit 8fa7e09 into apache:trunk Feb 20, 2026
22 checks passed

Comments

Conversation

eduwercamacaro commented Oct 22, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Nikita-Shupletsov left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

eduwercamacaro commented Dec 16, 2025

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

bbejeck commented Jan 9, 2026

Uh oh!

bbejeck commented Jan 13, 2026

Uh oh!

eduwercamacaro commented Feb 4, 2026

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

eduwercamacaro left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

bbejeck commented Feb 20, 2026

Uh oh!

omkreddy commented Feb 20, 2026

Uh oh!

bbejeck commented Feb 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

eduwercamacaro commented Oct 22, 2025 •

edited by github-actions bot

Loading