KAFKA-10500: Add thread

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -81,6 +81,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 import java.util.TreeMap;
@@ -155,6 +156,14 @@ public class KafkaStreams implements AutoCloseable {
     private final StreamsMetricsImpl streamsMetrics;
     private final ProcessorTopology taskTopology;
     private final ProcessorTopology globalTaskTopology;
+    private final StreamStateListener streamStateListener;
+    private final Map<Long, StreamThread.State> threadState;
+    private final ArrayList<StreamThreadStateStoreProvider> storeProviders;
+    private Long totalCacheSize;
+    private final UUID processId;
+    private final InternalTopologyBuilder internalTopologyBuilder;
+    private final KafkaClientSupplier clientSupplier;
+    private final DelegatingStateRestoreListener delegatingStateRestoreListener;
 
     GlobalStreamThread globalStreamThread;
     private KafkaStreams.StateListener stateListener;
@@ -660,18 +669,18 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
         this.config = config;
         this.time = time;
         // The application ID is a required config and hence should always have value
-        final UUID processId = UUID.randomUUID();
+        processId = UUID.randomUUID();
         final String userClientId = config.getString(StreamsConfig.CLIENT_ID_CONFIG);
         final String applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG);
         if (userClientId.length() <= 0) {
             clientId = applicationId + "-" + processId;
         } else {
             clientId = userClientId;
         }
-
+        this.internalTopologyBuilder = internalTopologyBuilder;
         final LogContext logContext = new LogContext(String.format("stream-client [%s] ", clientId));
         this.log = logContext.logger(getClass());
-
+        this.clientSupplier = clientSupplier;
         final MetricConfig metricConfig = new MetricConfig()
             .samples(config.getInt(StreamsConfig.METRICS_NUM_SAMPLES_CONFIG))
             .recordLevel(Sensor.RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
@@ -728,12 +737,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 = getCacheSizePerThread(numStreamThreads);
         final boolean hasPersistentStores = taskTopology.hasPersistentLocalStore() ||
                 (hasGlobalTopology && globalTaskTopology.hasPersistentGlobalStore());
 
@@ -743,7 +748,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
             throw new StreamsException(fatal);
         }
 
-        final StateRestoreListener delegatingStateRestoreListener = new DelegatingStateRestoreListener();
+        delegatingStateRestoreListener = new DelegatingStateRestoreListener();
         GlobalStreamThread.State globalThreadState = null;
         if (hasGlobalTopology) {
             final String globalThreadId = clientId + "-GlobalStreamThread";
@@ -764,8 +769,8 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
         // use client id instead of thread client id since this admin client may be shared among threads
         adminClient = clientSupplier.getAdmin(config.getAdminConfigs(ClientUtils.getSharedAdminClientId(clientId)));
 
-        final Map<Long, StreamThread.State> threadState = new HashMap<>(numStreamThreads);
-        final ArrayList<StreamThreadStateStoreProvider> storeProviders = new ArrayList<>();
+        threadState = new HashMap<>(numStreamThreads);
+        storeProviders = new ArrayList<>();
         for (int i = 0; i < numStreamThreads; i++) {
             final StreamThread streamThread = StreamThread.create(
                 internalTopologyBuilder,
@@ -789,7 +794,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
         ClientMetrics.addNumAliveStreamThreadMetric(streamsMetrics, (metricsConfig, now) ->
             Math.toIntExact(threads.stream().filter(thread -> thread.state().isAlive()).count()));
 
-        final StreamStateListener streamStateListener = new StreamStateListener(threadState, globalThreadState);
+        streamStateListener = new StreamStateListener(threadState, globalThreadState);
         if (hasGlobalTopology) {
             globalStreamThread.setStateListener(streamStateListener);
         }
@@ -806,6 +811,65 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
         rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, config);
     }
 
+    /**
+     * Adds and starts a stream thread in addition to the stream threads that are already running in this
+     * Kafka Streams client.
+     *
+     * Since the number of stream threads increases, the sizes of the caches in the new stream thread
+     * and the existing stream threads are adapted so that the sum of the cache sizes over all stream
+     * threads does not exceed the total cache size specified in configuration
+     * {@code cache.max.bytes.buffering}.
+     *
+     * Stream threads can only be added if this Kafka Streams client is in state RUNNING or REBALANCING.
+     *
+     * @return name of the added stream thread or empty if a new stream thread could not be added
+     */
+    public Optional<String> addStreamThread() {
+        synchronized (stateLock) {
+            if (state == State.RUNNING || state == State.REBALANCING) {
+                final int threadIdx = threads.size() + 1;
+                final long cacheSizePerThread = getCacheSizePerThread(threadIdx);
+                resizeThreadCache(threadIdx);
+                final StreamThread streamThread = StreamThread.create(
+                        internalTopologyBuilder,
+                        config,
+                        clientSupplier,
+                        adminClient,
+                        processId,
+                        clientId,
+                        streamsMetrics,
+                        time,
+                        streamsMetadataState,
+                        cacheSizePerThread,
+                        stateDirectory,
+                        delegatingStateRestoreListener,
+                        threadIdx);
+                threads.add(streamThread);
+                threadState.put(streamThread.getId(), streamThread.state());
+                storeProviders.add(new StreamThreadStateStoreProvider(streamThread));
+                streamThread.setStateListener(streamStateListener);
+                return Optional.of(streamThread.getName());
+            } else {
+                return Optional.empty();
+            }
+        }
+    }
+
+    private long getCacheSizePerThread(final int numStreamThreads) {
+        if (totalCacheSize < 0) {
+            totalCacheSize = 0L;
+            log.warn("Negative cache size passed in. Reverting to cache size of 0 bytes.");
+        }
+        return totalCacheSize / (numStreamThreads + ((globalTaskTopology != null) ? 1 : 0));
+    }
+
+    private void resizeThreadCache(final int numStreamThreads) {
+        final long cacheSizePreThread = getCacheSizePerThread(numStreamThreads);
+        for (final StreamThread streamThread: threads) {
+            streamThread.resizeCache(cacheSizePreThread);
+        }
+    }
+
     private ScheduledExecutorService setupStateDirCleaner() {
         return Executors.newSingleThreadScheduledExecutor(r -> {
             final Thread thread = new Thread(r, clientId + "-CleanupThread");

streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java

@@ -296,4 +296,7 @@ private LogContext getLogContext(final TaskId taskId) {
         return new LogContext(logPrefix);
     }
 
+    public void resizeCache(final long size) {
+        cache.resize(size);
+    }
 }

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -589,6 +589,10 @@ private void subscribeConsumer() {
         }
     }
 
+    public void resizeCache(final long size) {
+        taskManager.resizeCache(size);
+    }
+
     /**
      * One iteration of a thread includes the following steps:
      *

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

@@ -133,6 +133,10 @@ boolean isRebalanceInProgress() {
         return rebalanceInProgress;
     }
 
+    void resizeCache(final long size) {
+        activeTaskCreator.resizeCache(size);
+    }
+
     void handleRebalanceStart(final Set<String> subscribedTopics) {
         builder.addSubscribedTopicsFromMetadata(subscribedTopics, logPrefix);
 

streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java

@@ -35,7 +35,7 @@
  */
 public class ThreadCache {
     private final Logger log;
-    private final long maxCacheSizeBytes;
+    private long maxCacheSizeBytes;
     private final StreamsMetricsImpl metrics;
     private final Map<String, NamedCache> caches = new HashMap<>();
 
@@ -71,6 +71,16 @@ public long flushes() {
         return numFlushes;
     }
 
+    public void resize(final long maxCacheSizeBytes) {
+        final boolean shrink = maxCacheSizeBytes < this.maxCacheSizeBytes;
+        this.maxCacheSizeBytes = maxCacheSizeBytes;
+        if (shrink) {
+            for (final NamedCache cache : caches.values()) {
+                maybeEvict(cache.name());
+            }
+        }
+    }
+
     /**
      * The thread cache maintains a set of {@link NamedCache}s whose names are a concatenation of the task ID and the
      * underlying store name. This method creates those names.

streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java

@@ -471,6 +471,19 @@ public void shouldCalculateSizeInBytes() {
         assertEquals(cache.sizeBytes(), node.size());
     }
 
+    @Test
+    public void shouldResizeAndShrink() {
+        final ThreadCache cache = new ThreadCache(logContext, 10000, new MockStreamsMetrics(new Metrics()));
+        cache.put(namespace, Bytes.wrap(new byte[]{1}), cleanEntry(new byte[]{0}));
+        cache.put(namespace, Bytes.wrap(new byte[]{2}), cleanEntry(new byte[]{0}));
+        cache.put(namespace, Bytes.wrap(new byte[]{3}), cleanEntry(new byte[]{0}));
+        assertEquals(141, cache.sizeBytes());
+        cache.resize(100);
+        assertEquals(94, cache.sizeBytes());
+        cache.put(namespace1, Bytes.wrap(new byte[]{4}), cleanEntry(new byte[]{0}));
+        assertEquals(94, cache.sizeBytes());
+    }
+
     private LRUCacheEntry dirtyEntry(final byte[] key) {
         return new LRUCacheEntry(key, null, true, -1, -1, -1, "");
     }