muralibasani commented on code in PR #22458:
URL: https://github.com/apache/kafka/pull/22458#discussion_r3625087563
##########
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:
Agree. Went with 2nd option. A per-topology input.buffer.max.bytes override
now logs a WARN.
--
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]
