cadonna commented on a change in pull request #9695:
URL: https://github.com/apache/kafka/pull/9695#discussion_r553380512
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -947,24 +948,65 @@ private StreamThread createStreamThread(final long
cacheSizePerThread, final int
* @return name of the added stream thread or empty if a new stream thread
could not be added
*/
public Optional<String> addStreamThread() {
- synchronized (changeThreadCount) {
- if (isRunningOrRebalancing()) {
- final int threadIdx = getNextThreadIndex();
- final long cacheSizePerThread =
getCacheSizePerThread(threads.size() + 1);
+ if (isRunningOrRebalancing()) {
+ final int threadIdx;
+ final long cacheSizePerThread;
+ synchronized (changeThreadCount) {
+ threadIdx = getNextThreadIndex();
+ cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
resizeThreadCache(cacheSizePerThread);
- final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
- synchronized (stateLock) {
- if (isRunningOrRebalancing()) {
- streamThread.start();
- return Optional.of(streamThread.getName());
- } else {
+ }
+ final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
Review comment:
nit: Could you please rename the `createStreamThread()` to
`createAndAddStreamThread()`. It is a bit weird that we have `threads.remove()`
in this method but no `threads.add()`. The renaming would make it clearer.
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1432,7 +1480,9 @@ public void cleanUp() {
validateIsRunningOrRebalancing();
final Set<ThreadMetadata> threadMetadata = new HashSet<>();
for (final StreamThread thread : threads) {
- threadMetadata.add(thread.threadMetadata());
+ if (thread.state() != StreamThread.State.DEAD) {
+ threadMetadata.add(thread.threadMetadata());
+ }
}
return threadMetadata;
Review comment:
Do we not need to synchronize this block on the `changeThreadCount` to
at least guarantee that the state is correct when the method returns? Otherwise
between the `if` and `threadMetadata.add(thread.threadMetadata())` the stream
thread might transit to `DEAD`.
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -947,24 +948,65 @@ private StreamThread createStreamThread(final long
cacheSizePerThread, final int
* @return name of the added stream thread or empty if a new stream thread
could not be added
*/
public Optional<String> addStreamThread() {
- synchronized (changeThreadCount) {
- if (isRunningOrRebalancing()) {
- final int threadIdx = getNextThreadIndex();
- final long cacheSizePerThread =
getCacheSizePerThread(threads.size() + 1);
+ if (isRunningOrRebalancing()) {
+ final int threadIdx;
+ final long cacheSizePerThread;
+ synchronized (changeThreadCount) {
+ threadIdx = getNextThreadIndex();
+ cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
resizeThreadCache(cacheSizePerThread);
- final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
- synchronized (stateLock) {
- if (isRunningOrRebalancing()) {
- streamThread.start();
- return Optional.of(streamThread.getName());
- } else {
+ }
+ final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
+
+ synchronized (stateLock) {
+ if (isRunningOrRebalancing()) {
+ streamThread.start();
+ return Optional.of(streamThread.getName());
+ } else {
+ streamThread.shutdown();
+ threads.remove(streamThread);
+ resizeThreadCache(getCacheSizePerThread(threads.size()));
+ log.warn("Cannot add a stream thread in state " + state());
+ return Optional.empty();
+ }
+ }
+ }
+ log.warn("Cannot add a stream thread in state " + state());
+ return Optional.empty();
+ }
+
+ /**
+ * Removes one stream thread out of the running stream threads from this
Kafka Streams client.
+ * <p>
+ * The removed stream thread is gracefully shut down. This method does not
specify which stream
+ * thread is shut down.
+ * <p>
+ * Since the number of stream threads decreases, the sizes of the caches
in the remaining stream
+ * threads are adapted so that the sum of the cache sizes over all stream
threads equals the total
+ * cache size specified in configuration {@link
StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG}.
+ *
+ * @return name of the removed stream thread or empty if a stream thread
could not be removed because
+ * no stream threads are alive
+ */
+ public Optional<String> removeStreamThread() {
+ if (isRunningOrRebalancing()) {
+ synchronized (changeThreadCount) {
+ for (final StreamThread streamThread : threads) {
+ if (streamThread.isAlive() &&
(!streamThread.getName().equals(Thread.currentThread().getName()) ||
threads.size() == 1)) {
streamThread.shutdown();
+ if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
+
streamThread.waitOnThreadState(StreamThread.State.DEAD);
+ }
threads.remove(streamThread);
-
resizeThreadCache(getCacheSizePerThread(threads.size()));
- return Optional.empty();
+ final long cacheSizePerThread =
getCacheSizePerThread(threads.size());
+ resizeThreadCache(cacheSizePerThread);
+ return Optional.of(streamThread.getName());
}
}
}
+ log.warn("There are no threads eligible for removal");
+ } else {
+ log.warn("Cannot remove a stream thread in state " + state());
Review comment:
See my nit comment above.
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -947,24 +948,65 @@ private StreamThread createStreamThread(final long
cacheSizePerThread, final int
* @return name of the added stream thread or empty if a new stream thread
could not be added
*/
public Optional<String> addStreamThread() {
- synchronized (changeThreadCount) {
- if (isRunningOrRebalancing()) {
- final int threadIdx = getNextThreadIndex();
- final long cacheSizePerThread =
getCacheSizePerThread(threads.size() + 1);
+ if (isRunningOrRebalancing()) {
+ final int threadIdx;
+ final long cacheSizePerThread;
+ synchronized (changeThreadCount) {
+ threadIdx = getNextThreadIndex();
+ cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
resizeThreadCache(cacheSizePerThread);
- final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
- synchronized (stateLock) {
- if (isRunningOrRebalancing()) {
- streamThread.start();
- return Optional.of(streamThread.getName());
- } else {
+ }
+ final StreamThread streamThread =
createStreamThread(cacheSizePerThread, threadIdx);
+
+ synchronized (stateLock) {
+ if (isRunningOrRebalancing()) {
+ streamThread.start();
+ return Optional.of(streamThread.getName());
+ } else {
+ streamThread.shutdown();
+ threads.remove(streamThread);
+ resizeThreadCache(getCacheSizePerThread(threads.size()));
+ log.warn("Cannot add a stream thread in state " + state());
+ return Optional.empty();
Review comment:
nit: To deduplicate code, you could also remove these two lines, because
they are the same as the last two lines of the method.
nit: Could you change `Cannot add a stream thread in state " + state()` to
`Cannot add a stream thread when Kafka Streams client is in state " + state()`,
or similar. Currently, it is not completely clear if the state belongs to the
stream thread or to the client.
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -424,6 +424,7 @@ public void setUncaughtExceptionHandler(final
StreamsUncaughtExceptionHandler st
private void defaultStreamsUncaughtExceptionHandler(final Throwable
throwable) {
if (oldHandler) {
+ threads.remove((StreamThread) Thread.currentThread());
Review comment:
nit: My IDE says that the cast is not needed.
----------------------------------------------------------------
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:
us...@infra.apache.org
