From 232513429712f3dfc2654151313128e78668c233 Mon Sep 17 00:00:00 2001 From: Walker Carlson Date: Fri, 6 Nov 2020 10:41:56 -0800 Subject: [PATCH 1/5] ThreadCache Resizes --- .../apache/kafka/streams/KafkaStreams.java | 24 ++++++++++++++----- .../internals/ActiveTaskCreator.java | 3 +++ .../processor/internals/StreamThread.java | 4 ++++ .../processor/internals/TaskManager.java | 4 ++++ .../streams/state/internals/ThreadCache.java | 12 +++++++++- .../state/internals/ThreadCacheTest.java | 13 ++++++++++ 6 files changed, 53 insertions(+), 7 deletions(-) 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..9299e0f32137a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -155,6 +155,7 @@ public class KafkaStreams implements AutoCloseable { private final StreamsMetricsImpl streamsMetrics; private final ProcessorTopology taskTopology; private final ProcessorTopology globalTaskTopology; + private Long totalCacheSize; GlobalStreamThread globalStreamThread; private KafkaStreams.StateListener stateListener; @@ -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 = getCacheSizePreThread(numStreamThreads); final boolean hasPersistentStores = taskTopology.hasPersistentLocalStore() || (hasGlobalTopology && globalTaskTopology.hasPersistentGlobalStore()); @@ -806,6 +803,21 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder, rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, config); } + private long getCacheSizePreThread(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 = getCacheSizePreThread(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, ""); } From 68113b842861d85bda4616ad3d5c2f3a5e4adab9 Mon Sep 17 00:00:00 2001 From: Walker Carlson Date: Fri, 6 Nov 2020 15:08:21 -0800 Subject: [PATCH 2/5] fix typo --- .../main/java/org/apache/kafka/streams/KafkaStreams.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 9299e0f32137a..765ce3b4692f1 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -730,7 +730,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder, } totalCacheSize = config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG); - final long cacheSizePerThread = getCacheSizePreThread(numStreamThreads); + final long cacheSizePerThread = getCacheSizePerThread(numStreamThreads); final boolean hasPersistentStores = taskTopology.hasPersistentLocalStore() || (hasGlobalTopology && globalTaskTopology.hasPersistentGlobalStore()); @@ -803,7 +803,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder, rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, config); } - private long getCacheSizePreThread(final int numStreamThreads) { + 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."); @@ -812,7 +812,7 @@ private long getCacheSizePreThread(final int numStreamThreads) { } private void resizeThreadCache(final int numStreamThreads) { - final long cacheSizePreThread = getCacheSizePreThread(numStreamThreads); + final long cacheSizePreThread = getCacheSizePerThread(numStreamThreads); for (final StreamThread streamThread: threads) { streamThread.resizeCache(cacheSizePreThread); } From bb10405ee4dc5d2b4cb31b83313c71d10cefd208 Mon Sep 17 00:00:00 2001 From: Walker Carlson Date: Fri, 6 Nov 2020 15:06:52 -0800 Subject: [PATCH 3/5] add addThread --- .../apache/kafka/streams/KafkaStreams.java | 54 +++++++++++++++++-- 1 file changed, 50 insertions(+), 4 deletions(-) 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 765ce3b4692f1..aa6309890b584 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; @@ -156,6 +157,10 @@ public class KafkaStreams implements AutoCloseable { private final ProcessorTopology taskTopology; private final ProcessorTopology globalTaskTopology; 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; @@ -661,7 +666,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) { @@ -669,10 +674,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))) @@ -740,7 +745,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"; @@ -803,6 +808,47 @@ 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); + return Optional.of(streamThread.getName()); + } else { + return Optional.empty(); + } + } + } + private long getCacheSizePerThread(final int numStreamThreads) { if (totalCacheSize < 0) { totalCacheSize = 0L; From 6a0de6a28d29b13ea10fa1467fcd108be8da24bf Mon Sep 17 00:00:00 2001 From: Walker Carlson Date: Mon, 9 Nov 2020 15:43:42 -0800 Subject: [PATCH 4/5] resize after added, needs tests and number of threads metric --- .../src/main/java/org/apache/kafka/streams/KafkaStreams.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 aa6309890b584..4fe478ee664cc 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -826,7 +826,6 @@ public Optional addStreamThread() { 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, @@ -842,6 +841,7 @@ public Optional addStreamThread() { delegatingStateRestoreListener, threadIdx); threads.add(streamThread); + resizeThreadCache(threadIdx); return Optional.of(streamThread.getName()); } else { return Optional.empty(); From 66608e3e755d84c6fe80f15e662bdbe58cf851fc Mon Sep 17 00:00:00 2001 From: Walker Carlson Date: Tue, 10 Nov 2020 13:23:09 -0800 Subject: [PATCH 5/5] fixed state listeners --- .../org/apache/kafka/streams/KafkaStreams.java | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) 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 4fe478ee664cc..2d0abe7c6f4de 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -156,6 +156,9 @@ 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; @@ -766,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, @@ -791,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); } @@ -826,6 +829,7 @@ public Optional addStreamThread() { 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, @@ -841,7 +845,9 @@ public Optional addStreamThread() { delegatingStateRestoreListener, threadIdx); threads.add(streamThread); - resizeThreadCache(threadIdx); + threadState.put(streamThread.getId(), streamThread.state()); + storeProviders.add(new StreamThreadStateStoreProvider(streamThread)); + streamThread.setStateListener(streamStateListener); return Optional.of(streamThread.getName()); } else { return Optional.empty();