KAFKA-21230: Pause processing on uncommitted bytes by nicktelford · Pull Request #23728 · apache/kafka

nicktelford · 2026-10-06T13:34:40Z

During a long rebalance, a StreamThread can use native memory without
limit and fail with an out-of-memory error. When the uncommitted state
store bytes exceed statestore.uncommitted.max.bytes,
StreamThread#maybeCommit forces a commit. But TaskManager skips
commits during a rebalance, and runOnceWithoutProcessingThreads
continues to process records. Thus the uncommitted bytes continue to
increase. RocksDB keeps these bytes in a WriteBatch in native memory,
outside the block cache and the write buffers.

At the start of each iteration of the processing loop, StreamThread
now calls maybePauseProcessingForUncommittedBytes(), which checks
shouldCommitDueToUncommittedBytes(). Because maybeCommit commits at
the end of each iteration, the bytes exceed the limit at this point only
when that commit was skipped or failed. So the check needs no extra
state. While processing is paused, the thread does not process records
or run punctuators, because both write to the state stores. It continues
to poll, so the rebalance can complete, and it continues to call
maybeCommit. Processing resumes when a commit succeeds, or when a
commit on partition revocation flushes the buffers. The thread logs the
pause and the resume at INFO level.

This change decreases throughput during a long rebalance, but it
prevents the out-of-memory error. GlobalStreamThread does not need
this change, because one thread writes to its stores, checks the bytes
and flushes them, and nothing skips that flush. A separate pull request
will apply the same pause to processing threads.

Two new StreamThreadTest tests cover the change.
shouldNotProcessOrPunctuateWhileUncommittedBytesExceedLimit shows that
the thread polls and tries to commit, but does not process or punctuate,
while the commit is skipped.
shouldResumeProcessingOnceUncommittedBytesAreCommitted shows that
processing stays paused for the iteration in which the rebalance
completes, and resumes only after the commit succeeds.

🤖 Generated with Claude Code

Reviewers: Bill Bejeck bbejeck@apache.org

github-actions · 2026-10-15T03:21:19Z

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

bbejeck

Thanks @nicktelford LGTM - kicking off a system test and will merge after that passes

Copilot

🟡 Changes recommended

Initialization writes above the limit can indefinitely pause the thread because the affected task is not yet commit-eligible.

1 open issue

Prevent initialization buffer guard from pausing tasks indefinitely · New

What changed in this PR

This PR limits native-memory growth in Kafka Streams during prolonged rebalances by pausing processing when uncommitted state-store bytes exceed the configured limit.

Changes:

Skip record processing and punctuation above the limit while continuing polling and commit attempts.
Log transitions into and out of the paused state.
Add tests for skipped commits and resuming processing after a successful commit.

File	Description
`streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java`	Tests pause behavior and resumption after committing buffered bytes.
`streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java`	Adds the byte-limit processing guard and pause/resume logging.

_{Review effort: Balanced}

dismissing review waiting for co-pilot findings to be addressed

bbejeck · 2026-10-26T16:56:23Z

System tests pass

During a long rebalance, a StreamThread can use native memory without bound and fail with an out-of-memory error. When the uncommitted state store bytes exceed statestore.uncommitted.max.bytes, StreamThread#maybeCommit forces a commit. But TaskManager skips commits during a rebalance, and runOnceWithoutProcessingThreads continues to process records. Thus the uncommitted bytes, which RocksDB keeps in a WriteBatchWithIndex in native memory, continue to increase. At the start of each iteration of the processing loop, StreamThread now calls maybePauseProcessingForUncommittedBytes(), which checks shouldCommitDueToUncommittedBytes(). Because maybeCommit commits at the end of each iteration, the bytes exceed the limit at this point only when that commit was skipped or failed. While processing is paused, the thread does not process records or run punctuators, because both write to the state stores. It continues to poll and to call maybeCommit. Processing resumes when a commit succeeds, or when a commit on partition revocation flushes the buffers. The pause and the resume are logged at INFO level. Processor.init() can also write to transactional stores. Such a task has no records processed, so it is not eligible for commit, and the pause can hold it with nothing to process. When initialization of the topology leaves uncommitted bytes in the stores, StreamTask now marks the task as needing a commit, so that the next commit drains them. This change decreases throughput during a long rebalance, but prevents the out-of-memory error. A separate change will apply the same limit to processing threads. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Copilot

🟡 Changes recommended

Empty RocksDB batch overhead can keep processing paused indefinitely for valid small byte limits, even after successful commits.

2 open issues

_{Review effort: Balanced}

RocksDBTransactionBuffer.approximateNumUncommittedBytes() included the fixed header of an empty WriteBatch, which no commit releases. With a per-thread limit below the total of these headers, processing stayed paused after every commit. The empty header also made StreamTask request a commit after every restoration of a task with a transactional RocksDB store. The buffer now counts the WriteBatch only when it holds writes. The integration test is renamed to UncommittedBytesLimitIntegrationTest and gets a test with a one-byte limit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Copilot

🟢 Approval recommended

No unresolved blocking issues remain, and the regression coverage addresses both previously identified permanent-pause conditions; tests were not executed.

No open issues 🎉

2 resolved since last review

_{Review effort: Balanced}

bbejeck

Thanks for the fix @nicktelford LGTM.

I kicked off another system test run and will merge this once that completes.

bbejeck · 2026-10-27T16:01:20Z

2nd run of system tests pass

bbejeck · 2026-10-27T16:15:27Z

merged #23728 into trunk

During a long rebalance, a `StreamThread` can use native memory without limit and fail with an out-of-memory error. When the uncommitted state store bytes exceed `statestore.uncommitted.max.bytes`, `StreamThread#maybeCommit` forces a commit. But `TaskManager` skips commits during a rebalance, and `runOnceWithoutProcessingThreads` continues to process records. Thus the uncommitted bytes continue to increase. RocksDB keeps these bytes in a `WriteBatch` in native memory, outside the block cache and the write buffers. At the start of each iteration of the processing loop, `StreamThread` now calls `maybePauseProcessingForUncommittedBytes()`, which checks `shouldCommitDueToUncommittedBytes()`. Because `maybeCommit` commits at the end of each iteration, the bytes exceed the limit at this point only when that commit was skipped or failed. So the check needs no extra state. While processing is paused, the thread does not process records or run punctuators, because both write to the state stores. It continues to poll, so the rebalance can complete, and it continues to call `maybeCommit`. Processing resumes when a commit succeeds, or when a commit on partition revocation flushes the buffers. The thread logs the pause and the resume at INFO level. This change decreases throughput during a long rebalance, but it prevents the out-of-memory error. `GlobalStreamThread` does not need this change, because one thread writes to its stores, checks the bytes and flushes them, and nothing skips that flush. A separate pull request will apply the same pause to processing threads. Two new `StreamThreadTest` tests cover the change. `shouldNotProcessOrPunctuateWhileUncommittedBytesExceedLimit` shows that the thread polls and tries to commit, but does not process or punctuate, while the commit is skipped. `shouldResumeProcessingOnceUncommittedBytesAreCommitted` shows that processing stays paused for the iteration in which the rebalance completes, and resumes only after the commit succeeds. 4.4 backport: StreamThreadTest uses mock(TaskManager.class) instead of the no-arg mockTaskManager() helper, which only exists on trunk (#23313). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Reviewers: Bill Bejeck <bbejeck@apache.org> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

bbejeck · 2026-10-27T16:50:43Z

cherry-picked to 4.4

github-actions Bot added triage PRs from the community streams labels Oct 6, 2026

nicktelford force-pushed the KAFKA-21230 branch from 903c013 to b22b452 Compare October 6, 2026 13:46

github-actions Bot added the needs-attention label Oct 15, 2026

mjsax added ci-approved and removed triage PRs from the community needs-attention labels Oct 21, 2026

bbejeck previously approved these changes Oct 26, 2026

View reviewed changes

bbejeck requested a balanced review from Copilot October 26, 2026 15:28

Copilot started reviewing on behalf of bbejeck October 26, 2026 15:29

View session

Copilot AI reviewed Oct 26, 2026

View reviewed changes

Comment thread streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

nicktelford force-pushed the KAFKA-21230 branch from b22b452 to 458ef39 Compare October 26, 2026 18:40

bbejeck requested a balanced review from Copilot October 26, 2026 19:47

Copilot started reviewing on behalf of bbejeck October 26, 2026 19:48

View session

Copilot AI reviewed Oct 26, 2026

View reviewed changes

Comment thread streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

bbejeck requested a balanced review from Copilot October 27, 2026 12:57

Copilot started reviewing on behalf of bbejeck October 27, 2026 12:58

View session

Copilot AI reviewed Oct 27, 2026

View reviewed changes

bbejeck approved these changes Oct 27, 2026

View reviewed changes

bbejeck merged commit 0acc800 into apache:trunk Oct 27, 2026
24 checks passed

nicktelford mentioned this pull request Oct 27, 2026

KAFKA-21230: Pause task executors on byte limit #23984

Open

Navigation Menu

Repository navigation

KAFKA-21230: Pause processing on uncommitted bytes - #23728

KAFKA-21230: Pause processing on uncommitted bytes#23728
bbejeck merged 2 commits into
apache:trunkfrom
nicktelford:KAFKA-21230

nicktelford commented Oct 6, 2026 •

edited by github-actions Bot

Loading

Uh oh!

github-actions Bot commented Oct 15, 2026

Uh oh!

bbejeck left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

bbejeck commented Oct 26, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

bbejeck left a comment

Uh oh!

bbejeck commented Oct 27, 2026

Uh oh!

Uh oh!

bbejeck commented Oct 27, 2026

Uh oh!

bbejeck commented Oct 27, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Conversation

nicktelford commented Oct 6, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Oct 15, 2026

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

bbejeck commented Oct 26, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟢 Approval recommended

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

bbejeck commented Oct 27, 2026

Uh oh!

Uh oh!

bbejeck commented Oct 27, 2026

Uh oh!

bbejeck commented Oct 27, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

nicktelford commented Oct 6, 2026 •

edited by github-actions Bot

Loading