[GitHub] [kafka] mjsax commented on a change in pull request #9572: KAFKA-10500: Thread Cache Resizes

[GitBox <gi...@apache.org>]

mjsax commented on a change in pull request #9572:
URL: https://github.com/apache/kafka/pull/9572#discussion_r525399396



##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -728,12 +729,8 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
                 "must subscribe to at least one source topic or global table.");
         }
 
-        long totalCacheSize = config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG);
-        if (totalCacheSize < 0) {
-            totalCacheSize = 0;
-            log.warn("Negative cache size passed in. Reverting to cache size of 0 bytes.");
-        }
-        final long cacheSizePerThread = totalCacheSize / (numStreamThreads + (hasGlobalTopology ? 1 : 0));
+        totalCacheSize = config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG);
+        final long cacheSizePerThread = totalCacheSize / (numStreamThreads + ((globalTaskTopology != null) ? 1 : 0));

Review comment:
       Why move off using `hasGlobalTopology`?

##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -728,12 +729,8 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
                 "must subscribe to at least one source topic or global table.");
         }
 
-        long totalCacheSize = config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG);
-        if (totalCacheSize < 0) {
-            totalCacheSize = 0;

Review comment:
       Why do we remove this guard?

##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
##########
@@ -71,6 +72,26 @@ public long flushes() {
         return numFlushes;
     }
 
+    public void resize(final long maxCacheSizeBytes) {
+        final boolean shrink = maxCacheSizeBytes < this.maxCacheSizeBytes;
+        this.maxCacheSizeBytes = maxCacheSizeBytes;
+        if (shrink) {
+            final CircularIterator<NamedCache> circularIterator = new CircularIterator<>(caches.values());
+            while (sizeBytes() > maxCacheSizeBytes) {
+                if (!circularIterator.hasNext()) {
+                    log.error("Unable to remove any more entries as all caches are empty");

Review comment:
       Could this ever happen? If we the max cache size is smaller than a single entry, would we not evict the entry and the used cache size would always shrink to zero?

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -589,6 +593,10 @@ private void subscribeConsumer() {
         }
     }
 
+    public void resizeCache(final long size) {
+        cacheResizer.accept(size);

Review comment:
       Why do we need the `cacheResizer`? Can't we just call `cache.resize(size)` here?

##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
##########
@@ -71,6 +72,26 @@ public long flushes() {
         return numFlushes;
     }
 
+    public void resize(final long maxCacheSizeBytes) {

Review comment:
       nit: `newCachSizeBytes` ? (To avoid the "clash" with `this.maxCachSizeBytes`.)

##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -806,6 +803,13 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
         rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, config);
     }
 
+    private void resizeThreadCache(final int numStreamThreads) {
+        final long cacheSizePreThread = totalCacheSize / (numStreamThreads + ((globalTaskTopology != null) ? 1 : 0));

Review comment:
       Seems this duplicates `L733`. Might be good to extract into a small helper method.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org