frankvicky commented on code in PR #22458:
URL: https://github.com/apache/kafka/pull/22458#discussion_r3481815842
##########
streams/src/main/java/org/apache/kafka/streams/internals/StreamsConfigUtils.java:
##########
@@ -80,4 +82,46 @@ public static long totalCacheSize(final StreamsConfig
config) {
// only new or no config set. Use default or user specified value.
return config.getLong(STATESTORE_CACHE_MAX_BYTES_CONFIG);
}
+
+ // Returns -1 unless the deprecated config is the only one set; -1
disables the legacy per-partition pause.
+ @SuppressWarnings("deprecation")
+ public static int getBufferedRecordsPerPartition(final StreamsConfig
config) {
+ if
(config.originals().containsKey(BUFFERED_RECORDS_PER_PARTITION_CONFIG) &&
config.originals().containsKey(INPUT_BUFFER_MAX_BYTES_CONFIG)) {
+ LOG.warn("Both deprecated config {} and the new config {} are set,
hence {} is ignored and {} will be used instead to keep memory usage under
control.",
+ BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ INPUT_BUFFER_MAX_BYTES_CONFIG,
+ BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ INPUT_BUFFER_MAX_BYTES_CONFIG);
+ return -1;
+ } else if
(config.originals().containsKey(BUFFERED_RECORDS_PER_PARTITION_CONFIG)) {
+ LOG.warn("Deprecated config {} is set, and will be used; we
suggest setting the new config {} to keep memory usage under control " +
+ "instead as deprecated {} would be removed in the future.",
+ BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ INPUT_BUFFER_MAX_BYTES_CONFIG,
+ BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ return config.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ }
+ return -1;
+ }
+
+ // Returns -1L when only the legacy config is set; -1L disables the
bytes-based pause.
+ @SuppressWarnings("deprecation")
+ public static long getInputBufferMaxBytes(final StreamsConfig config) {
+ if
(config.originals().containsKey(BUFFERED_RECORDS_PER_PARTITION_CONFIG) &&
config.originals().containsKey(INPUT_BUFFER_MAX_BYTES_CONFIG)) {
+ LOG.warn("Both deprecated config {} and the new config {} are set,
hence {} is ignored and {} will be used instead to keep memory usage under
control.",
+ BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ INPUT_BUFFER_MAX_BYTES_CONFIG,
+ BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ INPUT_BUFFER_MAX_BYTES_CONFIG);
+ return config.getLong(INPUT_BUFFER_MAX_BYTES_CONFIG);
+ } else if
(config.originals().containsKey(BUFFERED_RECORDS_PER_PARTITION_CONFIG)) {
+ LOG.warn("Deprecated config {} is set, and will be used; we
suggest setting the new config {} to keep memory usage under control " +
+ "instead as deprecated {} would be removed in the future.",
+ BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ INPUT_BUFFER_MAX_BYTES_CONFIG,
+ BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ return -1L;
Review Comment:
This branch silently disables the new bytes-based cap for any user upgrading
from a release that had `buffered.records.per.partition` set, even though
`INPUT_BUFFER_MAX_BYTES_CONFIG` has a 512 MiB schema default that should still
apply.
Concrete migration path:
- User has been running 4.x with
`props.put(BUFFERED_RECORDS_PER_PARTITION_CONFIG, 2000);` (a common pattern for
memory tuning under bursty workloads).
- They upgrade to a build containing this PR; never touch the new key.
- `originals()` contains only the deprecated key → this branch fires →
returns `-1L`.
- `KafkaStreams.bufferSizePerThread` returns `-1L` for every thread →
`StreamThread.maxBufferSizeInBytes` stays `UNDEFINED` → bytes guard never runs.
KIP-770's stated goal — bounding memory for users on the deprecated config —
is precisely not achieved for the population the deprecation is aimed at. This
is structurally the same shape as the [Feb 2022 backward-compat
revert](https://github.com/apache/kafka/pull/11424) of the original KIP-770 PR
(the [resubmit](https://github.com/apache/kafka/pull/11796) introduced
`getTotalCacheSize()` exactly to avoid silently honoring only one side of the
old/new pair on the cache path; the input-buffer path needs the same treatment).
I think this branch should return the new config's value so the schema
default flows through:
```java
} else if
(config.originals().containsKey(BUFFERED_RECORDS_PER_PARTITION_CONFIG)) {
LOG.warn("Deprecated config {} is set; we suggest setting {} explicitly.
" +
"Using {} = {} (default) for memory bounding.",
BUFFERED_RECORDS_PER_PARTITION_CONFIG,
INPUT_BUFFER_MAX_BYTES_CONFIG,
INPUT_BUFFER_MAX_BYTES_CONFIG,
config.getLong(INPUT_BUFFER_MAX_BYTES_CONFIG));
return config.getLong(INPUT_BUFFER_MAX_BYTES_CONFIG);
}
```
The matching `getBufferedRecordsPerPartition` (lines 96-102) is correct
as-is: when only the deprecated config is set, honor it. The bytes helper just
shouldn't disable itself in that case.
If we keep the current behavior, please at least promote the log to `WARN`
and state explicitly that the new bytes cap is disabled — the current message
implies the opposite ("will be used... to keep memory usage under control").
##########
streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:
##########
@@ -312,6 +309,47 @@ 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 global input.buffer.max.bytes setting also locks out the legacy
per-topology override,
+ // since the user has explicitly opted into the bytes path app-wide.
+ final boolean inputBufferMaxBytesOverridden =
+ isTopologyOverride(INPUT_BUFFER_MAX_BYTES_CONFIG,
topologyOverrides)
+ ||
globalAppConfigs.originals().containsKey(INPUT_BUFFER_MAX_BYTES_CONFIG);
Review Comment:
Following up on the thread above — I agree with the spirit of @unknowntpo's
suggestion (a globally chosen bytes path should win over a topology-level
legacy override), but I think the current literal check is too loose and
reintroduces a silent-config-drop bug that's the same shape as the [Feb 2022
backward-compat revert](https://github.com/apache/kafka/pull/11424) of KIP-770.
`globalAppConfigs.originals().containsKey(INPUT_BUFFER_MAX_BYTES_CONFIG)`
returns true whenever the key appears in the user's `Properties` — even when
the value is identical to the schema default. So:
- NamedTopology user sets `topologyOverrides = {
buffered.records.per.partition: 1000 }` on topology A.
- Their global `Properties` contains `input.buffer.max.bytes = 536870912`
(the documented default, copied from a docs example or an ops template).
- `inputBufferMaxBytesOverridden` → true via `originals().containsKey(...)`.
- `setMaxBufferedRecordsPerPartition(true, true)` returns
`UNDEFINED_MAX_BUFFERED_SIZE` → the 1000-record topology override is silently
dropped with only an `INFO` log.
The user never opted into the bytes path; they just wrote down the
documented default.
Two ways to preserve @unknowntpo's intent without the silent drop:
**(a)** Detect "explicit opt-in" by comparing against the default:
```java
|| (globalAppConfigs.originals().containsKey(INPUT_BUFFER_MAX_BYTES_CONFIG)
&& globalAppConfigs.getLong(INPUT_BUFFER_MAX_BYTES_CONFIG)
!= /* default 512 MiB */)
```
Slightly fragile (a user could re-set to the default and expect lock-out),
but matches docs.
**(b)** Drop the global-side override-lock entirely. A topology-level
override is the strongest signal of intent; let it win. The two guards aren't
mutually exclusive — bytes-mode at `StreamThread` keeps the global memory
bound, count-mode at the affected topology keeps the per-partition fairness
bound. This is what the design already does when both configs are set at the
global level (`getBufferedRecordsPerPartition` returns `-1`,
`getInputBufferMaxBytes` returns the bytes value).
I lean (b): simpler, behavior matches user expectation ("if I asked for X on
topology A, I should get X on topology A"), and doesn't depend on default
detection. WDYT?
Whichever direction we go, if any silent-drop semantics remain I'd suggest
promoting the `INFO` log on line 334 to `WARN` — operators rarely spot-check
`INFO` for config behavior changes.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -804,9 +810,11 @@ record = partitionGroup.nextRecord(recordInfo,
wallClockTime);
log.trace("Task processed record: topic={}, partition={},
offset={}",
record.topic(), record.partition(), record.offset());
- // after processing this record, if its partition queue's buffered
size has been
- // decreased to the threshold, we can then resume the consumption
on this partition
- if (recordInfo.queue().size() <= maxBufferedSize) {
+ // headRecordIsCorrupted() catches a queue whose only remaining
records are corrupted —
+ // size() never drops below the legacy threshold, so without this
we'd stay paused forever.
+ if (recordInfo.queue().isEmpty()
Review Comment:
This branch un-pauses partitions that `StreamThread` paused for the bytes
guard, which I think re-introduces the perf-regression class that drove the Jul
2022 [revert](https://github.com/apache/kafka/pull/12383/) of KIP-770.
The pause side (line 1158, `addRecords`) is gated on `maxBufferedSize !=
UNDEFINED_MAX_BUFFERED_SIZE` — so in bytes-mode `StreamTask` never pauses. But
the resume side here fires on `recordInfo.queue().isEmpty()` (and
`headRecordIsCorrupted()`) regardless of mode. Combined with
`taskManager.resumePollingForPartitionsWithAvailableSpace()` at the top of
every `runOnce*` iteration, the consequence is:
1. `pollPhase` finds `totalBytes` > cap, pauses all non-empty partitions
`{P1..Pn}` on `mainConsumer` and records them in
`partitionsPausedForBufferOverflow`.
2. `StreamTask` processes some records on partition `Pa`; its queue empties;
`Pa` is added to `partitionsToResume` by this branch.
3. Next iteration: `mainConsumer.resume(Pa)` — even though `totalBytes`
across the other tasks is still well over the cap.
4. Consumer fetches a fresh batch on `Pa`; totalBytes spikes; goto 1.
End result: the per-thread bytes cap is enforced only as a soft jitter per
task, not as a true ceiling. Under sustained fan-in across multiple tasks this
becomes pause/resume churn on the consumer every iteration.
I think the resume conditions need to be gated by mode the same way the
pause side is, e.g.:
```java
if (maxBufferedSize != UNDEFINED_MAX_BUFFERED_SIZE
&& (recordInfo.queue().isEmpty()
|| recordInfo.queue().size() <= maxBufferedSize
|| recordInfo.queue().headRecordIsCorrupted())) {
partitionsToResume.add(partition);
}
```
— so `StreamTask` owns resume only for partitions `StreamTask` itself paused
(count-mode), and `StreamThread.maybeResumePartitionsPausedForBufferOverflow`
remains the sole owner of resume in bytes-mode.
The `headRecordIsCorrupted()` comment on line 813 explains the legacy
concern correctly, but that concern only applies when count-mode owns the pause
— in bytes-mode `StreamThread`'s threshold doesn't depend on `queue.size()`.
Would it be possible to add a test that exercises this: bytes cap N, two
tasks each with one partition, force one task to drain its queue while
totalBytes still exceeds N, and assert the consumer remains paused on its
partition?
--
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]
