You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/03/01 18:31:44 UTC
[GitHub] [kafka] guozhangwang commented on a change in pull request #11796: Kip 770
guozhangwang commented on a change in pull request #11796:
URL: https://github.com/apache/kafka/pull/11796#discussion_r817035430
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -116,14 +123,30 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
} else {
- maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ maxBufferedSize = globalAppConfigs.originals().containsKey(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG)
+ ? globalAppConfigs.getInt(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG) : -1;
}
- if (isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
- cacheSize = getLong(CACHE_MAX_BYTES_BUFFERING_CONFIG);
- log.info("Topology {} is overriding {} to {}", topologyName, CACHE_MAX_BYTES_BUFFERING_CONFIG, cacheSize);
+ if (isTopologyOverride(STATESTORE_CACHE_MAX_BYTES_CONFIG, topologyOverrides) ||
+ isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
+
+ if (isTopologyOverride(STATESTORE_CACHE_MAX_BYTES_CONFIG, topologyOverrides) && isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
+ cacheSize = getLong(STATESTORE_CACHE_MAX_BYTES_CONFIG);
+ log.info("Topology {} is using both {} and deprecated config {}. overriding {} to {}",
Review comment:
`overriding {} to {}` here could also be a bit confusing since it could be interpreted as `the second override the first` by some one. I'd suggest we just say `Both deprecated config {} and new config {} are set, hence {} is ignored and the new config {} (value {}) is used.`
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -116,14 +123,30 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
} else {
- maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ maxBufferedSize = globalAppConfigs.originals().containsKey(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG)
+ ? globalAppConfigs.getInt(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG) : -1;
}
- if (isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
- cacheSize = getLong(CACHE_MAX_BYTES_BUFFERING_CONFIG);
- log.info("Topology {} is overriding {} to {}", topologyName, CACHE_MAX_BYTES_BUFFERING_CONFIG, cacheSize);
+ if (isTopologyOverride(STATESTORE_CACHE_MAX_BYTES_CONFIG, topologyOverrides) ||
+ isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
+
+ if (isTopologyOverride(STATESTORE_CACHE_MAX_BYTES_CONFIG, topologyOverrides) && isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
+ cacheSize = getLong(STATESTORE_CACHE_MAX_BYTES_CONFIG);
+ log.info("Topology {} is using both {} and deprecated config {}. overriding {} to {}",
+ topologyName,
+ STATESTORE_CACHE_MAX_BYTES_CONFIG,
+ CACHE_MAX_BYTES_BUFFERING_CONFIG,
+ STATESTORE_CACHE_MAX_BYTES_CONFIG,
+ cacheSize);
+ } else if (isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
+ cacheSize = getLong(CACHE_MAX_BYTES_BUFFERING_CONFIG);
+ log.info("Topology {} is using deprecated config {}. overriding {} to {}", topologyName, CACHE_MAX_BYTES_BUFFERING_CONFIG, CACHE_MAX_BYTES_BUFFERING_CONFIG, cacheSize);
Review comment:
Ditto here: we just say "Only deprecated config {} is set and hence it's value {} would be used; we suggest setting the new config {} instead since the deprecated config {} would be removed in the future."
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -116,14 +123,30 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
} else {
- maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ maxBufferedSize = globalAppConfigs.originals().containsKey(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG)
+ ? globalAppConfigs.getInt(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG) : -1;
}
- if (isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
- cacheSize = getLong(CACHE_MAX_BYTES_BUFFERING_CONFIG);
- log.info("Topology {} is overriding {} to {}", topologyName, CACHE_MAX_BYTES_BUFFERING_CONFIG, cacheSize);
+ if (isTopologyOverride(STATESTORE_CACHE_MAX_BYTES_CONFIG, topologyOverrides) ||
+ isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
+
+ if (isTopologyOverride(STATESTORE_CACHE_MAX_BYTES_CONFIG, topologyOverrides) && isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
+ cacheSize = getLong(STATESTORE_CACHE_MAX_BYTES_CONFIG);
+ log.info("Topology {} is using both {} and deprecated config {}. overriding {} to {}",
+ topologyName,
+ STATESTORE_CACHE_MAX_BYTES_CONFIG,
+ CACHE_MAX_BYTES_BUFFERING_CONFIG,
+ STATESTORE_CACHE_MAX_BYTES_CONFIG,
+ cacheSize);
+ } else if (isTopologyOverride(CACHE_MAX_BYTES_BUFFERING_CONFIG, topologyOverrides)) {
+ cacheSize = getLong(CACHE_MAX_BYTES_BUFFERING_CONFIG);
+ log.info("Topology {} is using deprecated config {}. overriding {} to {}", topologyName, CACHE_MAX_BYTES_BUFFERING_CONFIG, CACHE_MAX_BYTES_BUFFERING_CONFIG, cacheSize);
+ } else {
+ cacheSize = getLong(STATESTORE_CACHE_MAX_BYTES_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, STATESTORE_CACHE_MAX_BYTES_CONFIG, cacheSize);
Review comment:
I think if just the new config is used (which is expected to be the norm in the future) we do not need to log anything.
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
##########
@@ -1407,6 +1430,26 @@ private void verifyMaxInFlightRequestPerConnection(final Object maxInFlightReque
return props;
}
+ public long getTotalCacheSize() {
+ // both deprecated and new config set. Warn and use the new one.
+ if (originals().containsKey(CACHE_MAX_BYTES_BUFFERING_CONFIG) && originals().containsKey(STATESTORE_CACHE_MAX_BYTES_CONFIG)) {
+ log.warn("Use of deprecated config {} noticed.", CACHE_MAX_BYTES_BUFFERING_CONFIG);
Review comment:
This warn log line would be confusing I'm afraid, since we print this line in both cases, whereas here we are not actually using the deprecated config, but the new one. I'd suggest we differentiate this line with the line 1445 below, e.g.:
* Remove this log line in 1436.
* Update log line in 1438:
```
Both deprecated config {} and the new config {} are set, hence {} is ignored and {} is used instead.
```
* Update log line in 1446 as:
```
Deprecated config {} is set, and will be used; we suggest setting the new config {} instead as deprecated {} would be removed in the future.
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org