frankvicky commented on code in PR #22458:
URL: https://github.com/apache/kafka/pull/22458#discussion_r4360732067
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1540,6 +1581,22 @@ 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).
+ // copy first — nonEmptyPartitions may be immutable, and pausing
an unassigned partition
+ // throws, so intersect with assignment (mirrors resume side).
+ final Set<TopicPartition> nonEmptyPartitions = new
HashSet<>(taskManager.nonEmptyPartitions());
+ nonEmptyPartitions.retainAll(mainConsumer.assignment());
+ if (!nonEmptyPartitions.isEmpty()) {
+ log.info("Buffered records size {} bytes exceeds {}. Pausing
partitions {} from the consumer",
Review Comment:
Pause and resume both log at INFO with the full partition list. Near the cap
this fires every iteration (pause at `> cap`, resume at `<= cap`, no
hysteresis), so it becomes two INFO lines per loop. Please move to DEBUG, same
for the resume log above.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1215,8 +1226,32 @@ 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);
+ }
+
+ // Drop revoked partitions so we never later resume one now owned by
another path (reset/rebalance).
+ void removePartitionsFromBufferOverflowTracking(final
Collection<TopicPartition> revokedPartitions) {
+ partitionsPausedForBufferOverflow.removeAll(revokedPartitions);
+ }
+
+ private void maybeResumePartitionsPausedForBufferOverflow() {
Review Comment:
This still resumes partitions owned by another pauser on the
`closeDirtyAndRevive` path. There the task's partitions stay in
`mainConsumer.assignment()` but the task is removed from
`activeTasksPerPartition`, and `closeDirty` clears its buffer so total bytes
drop below the cap. The next call here resumes them, the following poll returns
records for them, and `getActiveTask` throws NPE and kills the thread.
Revoke/lost pruning doesn't cover this. I'd suggest also intersecting
`toResume` with the input partitions of currently active-initialized tasks (a
small `TaskManager` helper), which closes this and any similar lifecycle gap in
one place.
##########
streams/src/main/java/org/apache/kafka/streams/internals/StreamsConfigUtils.java:
##########
@@ -85,4 +87,50 @@ 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;
+ }
+
+ // Always returns a usable bytes cap so memory stays bounded (KIP-770).
Only when the legacy
+ // config alone is set do we fall through to the new config's default; the
legacy per-partition
+ // record cap still applies in parallel via getBufferedRecordsPerPartition.
+ @SuppressWarnings("deprecation")
+ public static long getInputBufferMaxBytes(final StreamsConfig config) {
Review Comment:
The three branches return the same value and the "both set" warning is also
logged by `getBufferedRecordsPerPartition`, so users see it twice. Could be
reduced to one `if` for the deprecated-only warning.
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1456,8 +1467,19 @@ private long maxUncommittedBytesPerThread(final int
numStreamThreads) {
return divisor == 0 ? totalMax : totalMax / divisor;
}
- private void resizeThreadCache(final long cacheSizePerThread) {
- processStreamThread(thread -> thread.resizeCache(cacheSizePerThread));
+ // Pass through -1L unchanged so StreamThread can tell the bytes guard is
disabled.
+ private long bufferSizePerThread(final int numStreamThreads) {
+ if (inputBufferMaxBytes == -1L) {
Review Comment:
`getInputBufferMaxBytes` always returns the config value (`atLeast(0)`), so
this `-1` branch is unreachable and the comment is stale. Either drop it or
keep a real way to disable the guard.
##########
docs/streams/upgrade-guide.md:
##########
@@ -97,6 +97,8 @@ As part of
[KIP-1284](https://cwiki.apache.org/confluence/spaces/KAFKA/pages/406
[KIP-1356](https://cwiki.apache.org/confluence/spaces/KAFKA/pages/430408653/KIP-1356+Introduce+IQv2+for+headers-aware+state+stores)
adds four `@Evolving` IQv2 query types that return record headers from the
[header-aware state stores](#kip-1271-headers-aware-stores) introduced by
KIP-1271: `TimestampedKeyWithHeadersQuery`, `TimestampedRangeWithHeadersQuery`,
`TimestampedWindowKeyWithHeadersQuery`, and
`TimestampedWindowRangeWithHeadersQuery`. Their results carry headers as a
[`ReadOnlyRecord`](/{version}/javadoc/org/apache/kafka/streams/processor/api/ReadOnlyRecord.html)
— or, for the range and window forms, a closeable
[`ReadOnlyRecordIterator`](/{version}/javadoc/org/apache/kafka/streams/state/ReadOnlyRecordIterator.html).
See the [interactive queries
guide](/{version}/streams/developer-guide/interactive-queries/#header-aware-stores-interactive-queries)
for usage and behavior details.
+[KIP-770](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186878390)
is now fully implemented: config `input.buffer.max.bytes` (LONG, default 512
MB, topology-wide) is introduced and `buffered.records.per.partition` is
deprecated ([KAFKA-13152](https://issues.apache.org/jira/browse/KAFKA-13152)).
The total buffer is split evenly across stream threads (`input.buffer.max.bytes
/ num.stream.threads`); when a thread's tasks collectively exceed their slice,
the main consumer pauses every non-empty input partition on that thread and
resumes once the buffered bytes drop back below the cap. When
`input.buffer.max.bytes` is set, the new bytes-based topology-wide pause
supersedes the legacy per-partition record-count pause. If only the deprecated
`buffered.records.per.partition` is set, the bytes cap is still applied at its
default (512 MB) so memory stays bounded, while the legacy per-partition record
cap continues to apply in parallel; a deprecation warning is logged. Memor
y note: upgraders who never set `input.buffer.max.bytes` will reserve 512 MB
for input buffering by default — set it explicitly on memory-constrained
deployments.
Review Comment:
Two wording issues: "reserve 512 MB" is misleading, it's a cap, nothing is
allocated up front. And please state explicitly that when neither config is
set, the previous implicit 1000 records/partition limit no longer applies.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListener.java:
##########
@@ -86,6 +86,7 @@ public void onTasksRevoked(final
Set<StreamsRebalanceData.TaskId> tasks) {
if (streamThread.state() != StreamThread.State.PENDING_SHUTDOWN) {
streamThread.setState(StreamThread.State.PARTITIONS_REVOKED);
}
+
streamThread.removePartitionsFromBufferOverflowTracking(partitionsToRevoke);
Review Comment:
`onAllTasksLost` below should prune the tracking set too, for symmetry with
`onPartitionsLost` in the classic listener.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -463,8 +463,10 @@ public class StreamsConfig extends AbstractConfig {
@SuppressWarnings("WeakerAccess")
public static final String BOOTSTRAP_SERVERS_CONFIG =
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
- /** {@code buffered.records.per.partition} */
+ /** {@code buffered.records.per.partition}
+ * @deprecated Since 4.4. Use {@link #INPUT_BUFFER_MAX_BYTES_CONFIG
"input.buffer.max.bytes"} instead. */
Review Comment:
Trunk is 4.5.0-SNAPSHOT and the 4.4 branch has been cut without this PR, so
this should say 4.5, and the upgrade-guide paragraph should move to the 4.5.0
section.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1244,6 +1279,9 @@ void runOnceWithoutProcessingThreads() {
final long pollLatency;
taskManager.resumePollingForPartitionsWithAvailableSpace();
Review Comment:
When only the deprecated config is set, both guards are active and they
fight: `StreamTask.process` adds any partition with `queue.size() <=
maxBufferedSize` to `partitionsToResume`, and this call resumes them
unconditionally at the start of the next iteration, undoing the bytes-guard
pause from the previous `pollPhase`. Net effect: resume -> fetch -> pause every
iteration and the bytes cap never holds for exactly the migration population
the upgrade note describes.
The legacy resume path should skip partitions in
`partitionsPausedForBufferOverflow` (e.g. pass the set down, or filter in
`StreamTask.resumePollingForPartitionsWithAvailableSpace`).
--
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]
