diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java index 9c47125d4b880..2d0abe7c6f4de 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/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 threadState; + private final ArrayList 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,7 +669,7 @@ 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) { @@ -668,10 +677,10 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder, } 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 threadState = new HashMap<>(numStreamThreads); - final ArrayList 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 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"); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java index 012ff20f8b980..a196be58664c6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java +++ b/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); + } } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java index bab195e8aa0b3..2ec06ab935388 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java +++ b/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: * diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java index 583ecf2e2efa3..31cbacc3f4365 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java +++ b/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 subscribedTopics) { builder.addSubscribedTopicsFromMetadata(subscribedTopics, logPrefix); diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java index 594dc4616b6c0..8385eeb08dc71 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java +++ b/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 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. diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java index 84431ad11c2dd..77be8126d1076 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java +++ b/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, ""); }