muralibasani commented on code in PR #22458:
URL: https://github.com/apache/kafka/pull/22458#discussion_r3560146098
##########
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:
Agree with this one too. getInputBufferMaxBytes now returns the new config's
default when only the deprecated config is set. The deprecated per-partition
record cap still applies in parallel. Added StreamsConfigTest cases too.
--
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]
