You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "nicktelford (via GitHub)" <gi...@apache.org> on 2026/10/27 17:10:10 UTC
[PR] KAFKA-21230: Pause task executors on byte limit [kafka]
nicktelford opened a new pull request, #23984:
URL: https://github.com/apache/kafka/pull/23984
#23728 pauses processing and punctuation while the uncommitted state store bytes of a StreamThread exceed `statestore.uncommitted.max.bytes`. That pause is in `runOnceWithoutProcessingThreads`. When processing threads are enabled, task executors process records and run punctuators independently of the StreamThread loop, so the pause does not stop them. If a commit is skipped during a rebalance, or fails, the executors continue to write to the state stores. The uncommitted bytes then increase without bound, and the native memory of the transaction buffers increases with them.
`StreamThread#maybePauseProcessingForUncommittedBytes()` now also sets the pause on the processing threads, through the new `TaskManager#setProcessingThreadsPaused`. `runOnceWithProcessingThreads` calls it after `maybeCommit`. The scheduling `TaskManager` gains `setProcessingPaused` and `processingPaused`.
`DefaultTaskManager` keeps the flag in a volatile field for each StreamThread. It does not use `TaskExecutionMetadata`, because all StreamThreads share that object. While the flag is set, `canProgress` returns false, so `assignNextTask` assigns no task and `awaitProcessableTasks` blocks. When processing resumes, `DefaultTaskManager` signals the executors. While paused, `DefaultTaskExecutor` does not process or punctuate its current task. The task makes no progress, so the executor releases it through the existing path for such tasks.
`GlobalStreamThread` does not need this change. One thread writes, checks and flushes synchronously, and no condition skips the flush.
Both tests in `UncommittedBytesLimitIntegrationTest` now also run with processing threads enabled.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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-21230: Pause task executors on byte limit [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #23984:
URL: https://github.com/apache/kafka/pull/23984#issuecomment-6507110226
cherry-picked to 4.4
--
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-21230: Pause task executors on byte limit [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #23984:
URL: https://github.com/apache/kafka/pull/23984#discussion_r4403578974
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutorTest.java:
##########
@@ -134,6 +134,20 @@ public void shouldUnassignTaskWhenNotProgressing() {
assertNull(taskExecutor.currentTask());
}
+ @Test
+ public void shouldUnassignTaskWhileProcessingPaused() {
+ when(taskManager.processingPaused()).thenReturn(true);
+
Review Comment:
Fixed in e97363eed182ab9a9c6b417f68991e271cbad2fa
--
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-21230: Pause task executors on byte limit [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #23984:
URL: https://github.com/apache/kafka/pull/23984#issuecomment-6506545878
[System tests passed](https://confluent-open-source-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/trunk/2026-10-28--001.5d949d2b-22a5-4be1-973d-88ca6829e681--1793211481--nicktelford--KAFKA-21230-processing-threads--36bd135070/report.html)
--
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-21230: Pause task executors on byte limit [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #23984:
URL: https://github.com/apache/kafka/pull/23984#issuecomment-6506633135
Merged #23984 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] KAFKA-21230: Pause task executors on byte limit [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #23984:
URL: https://github.com/apache/kafka/pull/23984#discussion_r4403581069
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1875,6 +1875,15 @@ public void signalTaskExecutors() {
}
}
+ /**
+ * Pause or resume processing and punctuation on the processing threads, if they are enabled.
+ */
+ void setProcessingThreadsPaused(final boolean paused) {
Review Comment:
Fixed in ce6075106aea3bcabbfe9b61cad5d9b784bf9628
--
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-21230: Pause task executors on byte limit [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #23984:
URL: https://github.com/apache/kafka/pull/23984#issuecomment-6472726718
Thanks for the PR @nicktelford - overall this lgtm just a couple of minor comments
--
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-21230: Pause task executors on byte limit [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck merged PR #23984:
URL: https://github.com/apache/kafka/pull/23984
--
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-21230: Pause task executors on byte limit [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #23984:
URL: https://github.com/apache/kafka/pull/23984#discussion_r4393758443
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1461,6 +1461,10 @@ void runOnceWithProcessingThreads() {
taskManager.activeTaskIds(), taskManager.standbyTaskIds(), commitLatency);
}
}
+
+ // maybeCommit() commits when the uncommitted bytes exceed the limit, so they only exceed it here when that
+ // commit was skipped (during a rebalance) or failed, or when the task executors wrote more since the commit
+ maybePauseProcessingForUncommittedBytes();
Review Comment:
With the Streams group protocol, a status-only heartbeat such as `ASSIGNMENT_DELAYED` can mark the group not ready while retaining active tasks. The early return at line 1427 then bypasses both commits and this pause check, so executors can keep processing and punctuating without the byte limit taking effect. Call `maybePauseProcessingForUncommittedBytes()` before that early return too, and add a processing-thread test for a ready-to-not-ready transition with retained tasks.
--
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-21230: Pause task executors on byte limit [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #23984:
URL: https://github.com/apache/kafka/pull/23984#discussion_r4402121126
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutorTest.java:
##########
@@ -134,6 +134,20 @@ public void shouldUnassignTaskWhenNotProgressing() {
assertNull(taskExecutor.currentTask());
}
+ @Test
+ public void shouldUnassignTaskWhileProcessingPaused() {
+ when(taskManager.processingPaused()).thenReturn(true);
+
Review Comment:
nit: `canPunctuateTask` isn't stubbed here, so the mock returns false and the punctuate branch is never entered, paused or not. That means the `never().maybePunctuateStreamTime()` / `never().maybePunctuateSystemTime()` checks pass even if the `!processingPaused` check on punctuation is removed.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1875,6 +1875,15 @@ public void signalTaskExecutors() {
}
}
+ /**
+ * Pause or resume processing and punctuation on the processing threads, if they are enabled.
+ */
+ void setProcessingThreadsPaused(final boolean paused) {
Review Comment:
nit: No test covers this forwarding. `StreamThreadTest` mocks `TaskManager`, so if this call became a no-op, every test would still pass
--
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