frankvicky commented on code in PR #22458:
URL: https://github.com/apache/kafka/pull/22458#discussion_r3615036374
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1207,8 +1218,27 @@ private void subscribeConsumer() {
}
}
- public void resizeCache(final long size) {
- cacheResizeSize.set(size);
+ public void resizeCacheAndBufferMemory(final long cacheSize, final long
maxBufferSize) {
+ cacheResizeSize.set(cacheSize);
+ maxBufferSizeInBytes.set(maxBufferSize);
+ }
+
+ private void maybeResumePartitionsPausedForBufferOverflow() {
+ if (maxBufferSizeInBytes.get() == UNDEFINED_INPUT_BUFFER_MAX_BYTES) {
+ return;
+ }
+ if (partitionsPausedForBufferOverflow.isEmpty()) {
+ return;
+ }
+ if (taskManager.getInputBufferSizeInBytes() <=
maxBufferSizeInBytes.get()) {
+ // defensive copy — we clear the tracking set right after.
+ final Set<TopicPartition> toResume = new
HashSet<>(partitionsPausedForBufferOverflow);
+ toResume.retainAll(mainConsumer.assignment());
Review Comment:
`toResume.retainAll(mainConsumer.assignment())` filters the resume set by
consumer assignment only, but I don't think that's sufficient.
`partitionsPausedForBufferOverflow` is never pruned on task close/revoke — I
checked `handleRevocation`, `handleTaskMigrated`, and `completeShutdown` and
none touch the set; the only clear is inside this method's success branch. So
under sustained bytes pressure a stale entry for a `TopicPartition` that's
still in `mainConsumer.assignment()` but currently paused by another owner
(offset-reset, rebalance settle) can pass the filter and get resumed.
Concrete trigger:
1. Bytes overflow → `pollPhase` adds `P` to
`partitionsPausedForBufferOverflow`.
2. `P` starts hitting `TaskCorruptedException` →
`TaskManager.closeDirtyAndRevive` runs →
`StreamTask.addPartitionsForOffsetReset(P)` pauses `P` again (this time for
offset reset) and `tasks.removeTask(task)` drops the task.
3. `P` is still in `mainConsumer.assignment()` **and** still in
`partitionsPausedForBufferOverflow` (nothing removed it).
4. Buffer drains below cap on the remaining tasks → this branch fires →
`toResume.retainAll(...)` keeps `P` → `mainConsumer.resume(P)` fires while the
offset-reset owner still expects `P` paused.
5. Consumer fetches from the un-reset offset; the reset-then-replay
invariant is broken.
Two ways to close this:
**(a)** Intersect against currently-active input partitions in addition to
assignment:
```java
toResume.retainAll(mainConsumer.assignment());
toResume.retainAll(taskManager.activeRunningTaskInputPartitions());
```
So the bytes guard only resumes partitions currently owned by an
active-running task, not ones transitioning through offset-reset or restore.
**(b)** Prune `partitionsPausedForBufferOverflow` proactively in
`handleRevocation` and any task-close path, so stale entries don't accumulate
at all. This also caps set growth under sustained pressure.
I lean (b) — cleaner invariant, no cross-cutting reads from `TaskManager` at
resume time. WDYT?
##########
streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:
##########
@@ -314,6 +311,46 @@ public TopologyConfig(final String topologyName, final
StreamsConfig globalAppCo
.getOrDefault(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"false")));
}
+ // local sentinel mirroring StreamTask.UNDEFINED_MAX_BUFFERED_SIZE; -1
disables the legacy
+ // per-partition pause and lets the bytes guard own buffering.
+ private static final int UNDEFINED_MAX_BUFFERED_SIZE = -1;
+
+ private int configureMaxBufferedSize() {
+ final boolean bufferedRecordsPerPartitionOverridden =
isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides);
+ // A topology-level override is the strongest signal of intent, so we
let it win locally; the
+ // thread-wide bytes guard still bounds memory globally. We
deliberately do NOT lock this out
+ // based on a global input.buffer.max.bytes value (which could just be
the documented default).
+ final boolean inputBufferMaxBytesOverridden =
isTopologyOverride(INPUT_BUFFER_MAX_BYTES_CONFIG, topologyOverrides);
Review Comment:
`INPUT_BUFFER_MAX_BYTES_CONFIG` is checked by `isTopologyOverride` here but
it isn't registered in `TopologyConfig`'s `CONFIG` `ConfigDef` (lines 116-179).
`isTopologyOverride` returns true purely on
`topologyOverrides.containsKey(config)` (see line 384), so a topology-level
override of `input.buffer.max.bytes` flips this boolean and disables the legacy
per-partition guard for that topology — but the value itself is never consumed
at the topology level. The actual bytes cap always flows from the global
`StreamsConfig` via `KafkaStreams.bufferSizePerThread`.
Concrete scenario:
- Named-topology user sets `topologyOverrides = { input.buffer.max.bytes:
100_000_000 }` on topology A, hoping to cap that topology at 100 MB.
- `inputBufferMaxBytesOverridden` → true.
- `setMaxBufferedRecordsPerPartition(false, true)` returns
`UNDEFINED_MAX_BUFFERED_SIZE`.
- topology A's `maxBufferedSize` is `-1` (legacy guard disabled) as expected.
- But the 100 MB value is silently discarded; topology A is effectively
bounded by the global `input.buffer.max.bytes` (default 512 MB) with no warning
that the per-topology override was ignored.
Two directions:
**(1)** Support per-topology bytes overrides properly — register
`INPUT_BUFFER_MAX_BYTES_CONFIG` in the `CONFIG` `ConfigDef` (like the existing
entry for `STATESTORE_CACHE_MAX_BYTES_CONFIG` at line 134), read the override
value in `configureMaxBufferedSize`, and thread it through to per-thread
accounting. This is a bigger change and is probably out of scope for this PR.
**(2)** Explicitly reject the override — if
`isTopologyOverride(INPUT_BUFFER_MAX_BYTES_CONFIG, ...)` is true, log a `WARN`
explaining that per-topology bytes caps aren't supported yet and the value is
being ignored, then treat it as if the override wasn't there (don't let it
disable the legacy guard either). This is a small fix.
I'd suggest (2) for this PR to avoid the silent-discard footgun, and file a
follow-up JIRA for (1) if there's actual demand. Either way, leaving it in the
current state where the override does one visible thing (disables legacy) but
not what the user asked for feels worse than either extreme.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1288,6 +1318,7 @@ void runOnceWithoutProcessingThreads() {
final int processed = taskManager.process(numIterations, time);
final long processLatency = advanceNowAndComputeLatency();
totalProcessLatency += processLatency;
+ maybeResumePartitionsPausedForBufferOverflow();
Review Comment:
There's an asymmetry between the two `runOnce` variants that I think leaves
the non-processing-threads path with a starvation window:
- `runOnceWithProcessingThreads` (line 1427): calls
`maybeResumePartitionsPausedForBufferOverflow()` at iteration start,
unconditionally.
- `runOnceWithoutProcessingThreads` (this line): only calls it inside `do {
... } while (true);` at line 1313+, and that `do/while` is gated by `if
(isStartingRunningOrPartitionAssigned())` on line 1299.
When the thread is in `PARTITIONS_REVOKED` or `PENDING_SHUTDOWN`
(transiently or otherwise) and the buffer has already drained below the cap
through commits / task suspend / etc., the resume path is never reached. The
consumer stays paused on the partitions we recorded earlier; the next iteration
returns 0 records; we loop waiting for something to re-open the do/while — or
for another bytes-overflow-then-drain cycle to reach the resume call.
Suggested fix: mirror the processing-threads variant and call
`maybeResumePartitionsPausedForBufferOverflow()` at the top of
`runOnceWithoutProcessingThreads` too. The method is already a no-op when
`partitionsPausedForBufferOverflow.isEmpty()` or when
`taskManager.getInputBufferSizeInBytes() > maxBufferSizeInBytes.get()`, so the
extra call is cheap and idempotent.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1526,6 +1559,17 @@ private long pollPhase() {
taskManager.updateNextOffsets(records.nextOffsets());
}
+ // soft cap — read isn't atomic across tasks, so small overshoots are
fine.
+ final long bufferSize = taskManager.getInputBufferSizeInBytes();
+ if (maxBufferSizeInBytes.get() != UNDEFINED_INPUT_BUFFER_MAX_BYTES &&
bufferSize > maxBufferSizeInBytes.get()) {
+ // pause only non-empty partitions; pausing empty ones risks
ordering deadlock (KAFKA-13152).
+ final Set<TopicPartition> nonEmptyPartitions =
taskManager.nonEmptyPartitions();
+ log.info("Buffered records size {} bytes exceeds {}. Pausing
partitions {} from the consumer",
+ bufferSize, maxBufferSizeInBytes.get(), nonEmptyPartitions);
+ mainConsumer.pause(nonEmptyPartitions);
Review Comment:
`mainConsumer.pause(nonEmptyPartitions)` here doesn't intersect with
`mainConsumer.assignment()`, unlike the symmetric resume path at line 1236
which was hardened with `toResume.retainAll(mainConsumer.assignment())` in
[2ba189433f](https://github.com/apache/kafka/pull/22458/commits/2ba189433f)
(via @unknowntpo's earlier feedback). The pause side is the mirror position and
I think needs the same guard — otherwise a `TopicPartition` that has just left
the consumer assignment during a rebalance can still be reported by
`taskManager.nonEmptyPartitions()` (its `PartitionGroup` still holds buffered
records because the task hasn't been suspended yet) and passed to
`mainConsumer.pause(...)`, which will throw `IllegalStateException("No current
assignment for partition <TP>")` from `SubscriptionState.assignedState` and
take the `StreamThread` down.
The trigger window is narrow (single-threaded `StreamThread` flow tends to
run `handleRevocation` inside the same `poll()`), but it's exactly the same
class of risk the resume-side fix already acknowledged, so keeping the two
sides symmetric feels safer than relying on the timing not to hit us.
Suggested minimal fix:
```java
if (maxBufferSizeInBytes.get() != UNDEFINED_INPUT_BUFFER_MAX_BYTES &&
bufferSize > maxBufferSizeInBytes.get()) {
// pause only non-empty partitions; pausing empty ones risks ordering
deadlock (KAFKA-13152).
final Set<TopicPartition> nonEmptyPartitions =
taskManager.nonEmptyPartitions();
nonEmptyPartitions.retainAll(mainConsumer.assignment());
if (nonEmptyPartitions.isEmpty()) {
return pollLatency;
}
log.info("Buffered records size {} bytes exceeds {}. Pausing partitions
{} from the consumer",
bufferSize, maxBufferSizeInBytes.get(), nonEmptyPartitions);
mainConsumer.pause(nonEmptyPartitions);
partitionsPausedForBufferOverflow.addAll(nonEmptyPartitions);
}
```
(Or extract the filtering into a small helper shared with the resume path,
since it's the same pattern.) An assertion / test with a
revoked-but-still-buffered `TopicPartition` would nail this in place — happy to
sketch one if useful.
--
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]
