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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
