muralibasani commented on code in PR #22458:
URL: https://github.com/apache/kafka/pull/22458#discussion_r3560138958
##########
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:
Agree. Went with option (b). Dropped the global-config lock. A
topology-level buffered.records.per.partition override now wins locally, while
the thread-wide bytes guard still bounds memory globally. This will avoid that
silent drop. Updated tests too. Basically this clause `||
globalAppConfigs.originals().containsKey(INPUT_BUFFER_MAX_BYTES_CONFIG)` is
removed now.
--
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]
