You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/11/19 17:54:35 UTC
[GitHub] [kafka] cadonna commented on a change in pull request #9615: KAFKA-10500: Add thread option
cadonna commented on a change in pull request #9615:
URL: https://github.com/apache/kafka/pull/9615#discussion_r527055689
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -894,11 +904,57 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
queryableStoreProvider = new QueryableStoreProvider(storeProviders, globalStateStoreProvider);
stateDirCleaner = setupStateDirCleaner();
- oldHandler = false;
maybeWarnAboutCodeInRocksDBConfigSetter(log, config);
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);
Review comment:
This should be:
```suggestion
final long cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
```
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -894,11 +904,57 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
queryableStoreProvider = new QueryableStoreProvider(storeProviders, globalStateStoreProvider);
stateDirCleaner = setupStateDirCleaner();
- oldHandler = false;
maybeWarnAboutCodeInRocksDBConfigSetter(log, config);
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;
Review comment:
Assume the following thread list [t2, t3, t4], `threadIdx` would be 4, which is already there. You should keep the currently used `threadIdx`s and check those to decide on the next `threadIdx`.
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -894,11 +904,57 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
queryableStoreProvider = new QueryableStoreProvider(storeProviders, globalStateStoreProvider);
stateDirCleaner = setupStateDirCleaner();
- oldHandler = false;
maybeWarnAboutCodeInRocksDBConfigSetter(log, config);
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,
+ KafkaStreams.this::closeToError,
+ streamsUncaughtExceptionHandler
+ );
+ 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();
+ }
+ }
+ }
Review comment:
Where is the stream thread started?
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -894,11 +904,57 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
queryableStoreProvider = new QueryableStoreProvider(storeProviders, globalStateStoreProvider);
stateDirCleaner = setupStateDirCleaner();
- oldHandler = false;
maybeWarnAboutCodeInRocksDBConfigSetter(log, config);
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);
Review comment:
```suggestion
resizeThreadCache(threads.size() + 1);
```
----------------------------------------------------------------
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