KAFKA-10317: Global thread should honor shutdown signal during bootstrapping by lucliu1108 · Pull Request #22417 · apache/kafka

lucliu1108 · 2026-05-29T17:37:52Z

Summary

This PR introduces a shutdown-aware bootstrap loop in GlobalStateManagerImpl and a consumer.wakeup() call during GlobalStreamThread.shutdown() that together let KafkaStreams#close() interrupt global-store restoration in progress, instead of waiting for the entire changelog to be replayed.

Ticket: https://issues.apache.org/jira/browse/KAFKA-10317

Implementation

The global thread passes its inErrorState() predicate to the state manager, which checks it before each batch in the bootstrap poll loop and exits cleanly when shutdown is requested. The wakeup() call additionally interrupts any in-flight poll() so shutdown takes effect right away, even if the loop is currently blocked on a fetch. A matching WakeupException catch in the main update loop ensures clean shutdowns aren't reported through the uncaught-exception handler.

Tests

Added unit tests in GlobalStateManagerImplTest covering the supplier check and WakeupException handling in both restoreState and reprocessState, and end-to-end tests in GlobalStreamThreadTest for the close-during-bootstrap scenario.

chickenchickenlove

Thanks for your hard work!
I left a comment.
When you get a chance, please take a look 🙇‍♂️

chickenchickenlove · 2026-05-30T05:14:29Z

+                if (inErrorStateSupplier.getAsBoolean()) {
+                    logBootstrapInterrupted(storeMetadata);
+                    return;


Could we make the shutdown-interrupted bootstrap path explicit instead of returning normally from GlobalStateManagerImpl?

Currently, when inErrorStateSupplier.getAsBoolean() is true, restoreState() / reprocessState() just return, so GlobalStateManagerImpl#initialize() can also return as if bootstrap completed successfully. As a result, GlobalStateUpdateTask#initialize() may continue into initTopology(), processorContext.initialize(), and flushState() even though shutdown has already been requested.

Since initTopology() can invoke user-provided Processor#init(), this could unnecessarily open external resources during shutdown. Maybe this should use an explicit internal signal, such as a dedicated bootstrap-interrupted exception caught only on the clean shutdown path, or return an initialize status like completed/interrupted so the follow-up initialization can be skipped.

What do you think?

Hi @chickenchickenlove , thanks for the review!
Good point, i refactored to make the interrupted path explicit by:

GlobalStateManager.initialize() now returns Optional<Set<String>> intead of Set<String>. Optional.empty() is the explicit "bootstrap was interrupted by shutdown" signal — set both when the supplier-check fires between polls and when a WakeupException is caught during shutdown.

GlobalStateUpdateTask.initialize() checks the Optional first; if empty, it returns Collections.emptyMap() immediately and skips the rest of the process.

UladzislauBlok

Made initial pass

UladzislauBlok · 2026-05-31T11:40:16Z

+                if (inErrorStateSupplier.getAsBoolean()) {
+                    logBootstrapInterrupted(storeMetadata);
+                    return;
+                }


Looks like a lot of code duplication. Can we move it to dedicated method, or keep it on high level?
AFAIU, when restoration will be completed (on current store) and we'll move to next one we'll interrupt it anyway. Kinda trade-off to not check same condition n-times

for (final StateStoreMetadata metadata : storeMetadata.values()) { if (inErrorStateSupplier.getAsBoolean()) { log.info("Global store bootstrap interrupted by shutdown before starting {}", metadata.stateStore.name()); break; } ... }

Thanks for the advice!
I have removed the check inside the restoreState or reprocessState and instead just have outer supplier check at per-store look in GlobalStateManagerImpl.initialize(), returning empty directly instead of continuing to the next store.

UladzislauBlok · 2026-05-31T11:51:15Z

+                try {
+                    stateConsumer.pollAndUpdate();
+                } catch (final WakeupException e) {
+                    if (!inErrorState()) {
+                        throw e;
+                    }
+                }


I think this is not part of bootstrapping, is it?

Good catch! The steady-state catch wasn't really about bootstrap. I removed it, and also tightened shutdown() to only triggering globalConsumer.wakeup() when the thread is still in CREATED state (i.e., still bootstrapping). The steady-state main loop is now untouched by this PR.

UladzislauBlok · 2026-05-31T11:55:27Z

+            if (inErrorState()) {
+                closeStateConsumer(stateConsumer, false);
+                return null;
+            }
+
            setState(RUNNING);
            return stateConsumer;
+        } catch (final WakeupException e) {
+            closeStateConsumer(stateConsumer, false);
+            if (inErrorState()) {
+                log.info("Global thread initialization interrupted by shutdown");
+            } else {
+                startupException = new StreamsException(
+                    "Unexpected wakeup during initialization of GlobalStreamThread", e);
+            }


I think this part should be enough:

if (inErrorState()) { closeStateConsumer(stateConsumer, false); return null; }

Do we need to catch WakeupException?
UPD: Overall idea is to break execution of GlobalStateManagerImpl and verify if it was interrupted (check inErrorState())

Thanks for flagging this!

For the catch {} block, it was originally used for covering the other blocking consumer calls during bootstrap that aren't inside the local poll-catch: GlobalStateManagerImpl#partitionsFor(), endOffsets() and position() (also part of the bootstrapping path). If shutdown() is fired during these paths, WakeupException` will propagate up directly.

Right now after applying the suggestion of @chickenchickenlove , this catch is no longer needed and has been removed. All WakeupException during bootstrap are now caught inside GlobalStateManagerImpl.initialize() and converted to Optional.empty(), so no WakeupExceiption reaches the GlobalStreamThread.initialize().

UladzislauBlok

Left minor comments, but LGTM overall

UladzislauBlok · 2026-06-03T16:19:26Z

+            LegacyCheckpointingStateStore.migrateLegacyOffsets(logPrefix, stateDirectory, null, wrappedStores);
+
+            for (final StateStoreMetadata metadata : storeMetadata.values()) {
+                if (shouldStopBootstrappingSupplier.getAsBoolean()) {


Just to confirm: so we ether catch the "interrupted bootstrapping" fact on next iteration or when handling WakeupException. That makes sense

Yes. First path is supplier check between stores, the other one is wakeupException thrown by methods like poll/partitionsFor/...

UladzislauBlok · 2026-06-03T16:21:35Z

+            if (inErrorState()) {
+                closeStateConsumer(stateConsumer, false);
+                return null;
+            }


for which scenario we need this check?

The previous try{ stateConsumer.initialize() } catch(...) block calls GlobalStateUpdateTask and will return directly if the shutdown signal is called. In that case, stateConsumer.initialize() returns normally with no exceptions thrown.

This follow-up checks the situation where shutdown is already requested, and routes to cleanup, causes the run() loop to go to the early-exit path.

right, I see it now. thanks for clarification

UladzislauBlok · 2026-06-03T16:28:39Z

+        final Thread shutdownThread = new Thread(() -> {
+            try {
+                TestUtils.waitForCondition(
+                    () -> stateRestoreListener.storeNameCalledStates.containsKey(MockStateRestoreListener.RESTORE_START),
+                    10 * 1000L,
+                    "Bootstrap restore never started.");
+            } catch (final Exception e) {
+                throw new RuntimeException(e);
+            }
+            globalStreamThread.shutdown();
+        });
+        shutdownThread.start();


Runnable + Executors is better imo, but this is super minor. You can ignore this comment if you like current approach more

Good point. Seems that the raw thread path would not surface the exceptions thrown inside the lambda to the main test thread and returns normally on shutdownThead.join(). I switched to the executor + runnable path.

UladzislauBlok · 2026-06-03T16:30:57Z

+
+        startAndSwallowError();
+        shutdownThread.join();
+        globalStreamThread.join(5_000);


there is already timeout on test, so we can remove this one I guess

Thanks! Removed.

lucliu1108 added 4 commits May 28, 2026 16:52

add supplier to catch shutdown signal while bootstrapping

7c26aca

Merge branch 'apache:trunk' into KAFKA-10317

1902941

cleanup

627150c

remove trailing spaces

bf57fef

github-actions Bot added triage PRs from the community streams labels May 29, 2026

chickenchickenlove reviewed May 30, 2026

View reviewed changes

github-actions Bot removed the triage PRs from the community label May 31, 2026

UladzislauBlok reviewed May 31, 2026

View reviewed changes

lucliu1108 added 3 commits June 1, 2026 22:18

Merge branch 'trunk' into KAFKA-10317

567b610

revise what state manager return when bootstrapping stopped

56abb15

remove check inside restoreState and reprocessState

7bddbf1

lucliu1108 requested review from UladzislauBlok and chickenchickenlove June 2, 2026 22:15

UladzislauBlok approved these changes Jun 3, 2026

View reviewed changes

revise tests

09935c5

Conversation

lucliu1108 commented May 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Implementation

Tests

Uh oh!

chickenchickenlove 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!

UladzislauBlok 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!

UladzislauBlok May 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

UladzislauBlok 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!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

lucliu1108 commented May 29, 2026 •

edited

Loading

UladzislauBlok May 31, 2026 •

edited

Loading