nicktelford commented on code in PR #22597:
URL: https://github.com/apache/kafka/pull/22597#discussion_r3426527947
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1324,13 +1333,29 @@ private long cacheSizePerThread(final int
numStreamThreads) {
return totalCacheSize / (numStreamThreads +
(topologyMetadata.hasGlobalTopology() ? 1 : 0));
}
+ private long maxUncommittedBytesPerThread(final int numStreamThreads) {
+ final long totalMax =
applicationConfigs.getLong(StreamsConfig.STATESTORE_UNCOMMITTED_MAX_BYTES_CONFIG);
+ if (totalMax <= 0) {
+ return -1;
+ }
+ final int divisor = Math.max(numStreamThreads, 0) +
(topologyMetadata.hasGlobalTopology() ? 1 : 0);
Review Comment:
It may seem strange to floor `numStreamThreads` at `0`, since
`StreamsConfig.NUM_STREAM_THREADS` has a minimum of `1`, but it's actually
possible for there to be 0 stream threads... If you `removeStreamThread` the
last remaining stream thread; or if you `replaceStreamThread`, there's a narrow
window where the old thread has been removed and the new one has not registered
yet.
--
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]
