ableegoldman commented on a change in pull request #9984:
URL: https://github.com/apache/kafka/pull/9984#discussion_r565611900
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -91,6 +93,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
Review comment:
There's actually a kafka-specific version of `TimeoutException` that you
should use to keep in line with other kafka APIs. It's
`org.apache.kafka.common.errors.TimeoutException`
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1005,11 +1008,60 @@ private StreamThread createAndAddStreamThread(final
long cacheSizePerThread, fin
|| threads.size() == 1)) {
streamThread.shutdown();
if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
-
streamThread.waitOnThreadState(StreamThread.State.DEAD);
+
streamThread.waitOnThreadState(StreamThread.State.DEAD, -1);
}
threads.remove(streamThread);
final long cacheSizePerThread =
getCacheSizePerThread(threads.size());
resizeThreadCache(cacheSizePerThread);
+ final Collection<MemberToRemove> membersToRemove =
Collections.singletonList(new
MemberToRemove(streamThread.getGroupInstanceID()));
+
adminClient.removeMembersFromConsumerGroup(config.getString(StreamsConfig.APPLICATION_ID_CONFIG),
new RemoveMembersFromConsumerGroupOptions(membersToRemove));
+ return Optional.of(streamThread.getName());
+ }
+ }
+ }
+ log.warn("There are no threads eligible for removal");
+ } else {
+ log.warn("Cannot remove a stream thread when Kafka Streams client
is 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}.
+ *
+ * @param timeout The the length of time to wait for the thread to shutdown
+ * @throws TimeoutException if the thread does not stop in time
+ * @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(final Duration timeout) throws
TimeoutException {
+ final String msgPrefix = prepareMillisCheckFailMsgPrefix(timeout,
"timeout");
+ final long timeoutMs = validateMillisecondDuration(timeout, msgPrefix);
+ if (isRunningOrRebalancing()) {
+ synchronized (changeThreadCount) {
+ // make a copy of threads to avoid holding lock
+ for (final StreamThread streamThread : new
ArrayList<>(threads)) {
+ if (streamThread.isAlive() &&
(!streamThread.getName().equals(Thread.currentThread().getName())
+ || threads.size() == 1)) {
+ streamThread.shutdown();
+ if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
+ if
(!streamThread.waitOnThreadState(StreamThread.State.DEAD, timeoutMs)) {
+ log.warn("Thread " + streamThread.getName() +
" did not stop in the allotted time");
+ throw new TimeoutException("Thread " +
streamThread.getName() + " did not stop in the allotted time");
+ }
+ }
+ threads.remove(streamThread);
+ final long cacheSizePerThread =
getCacheSizePerThread(threads.size());
+ resizeThreadCache(cacheSizePerThread);
+ Collection<MemberToRemove> membersToRemove =
Collections.singletonList(new
MemberToRemove(streamThread.getGroupInstanceID()));
Review comment:
I'm not sure how `removeMembersFromConsumerGroup` would behave if you
passed in `""` as the `group.instance.id`, do you know? If not then let's just
be safe and check what `streamThread.getGroupInstanceID()` returns, and skip
this call if there is no group.instance.id (ie if not static)
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1005,11 +1008,60 @@ private StreamThread createAndAddStreamThread(final
long cacheSizePerThread, fin
|| threads.size() == 1)) {
streamThread.shutdown();
if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
-
streamThread.waitOnThreadState(StreamThread.State.DEAD);
+
streamThread.waitOnThreadState(StreamThread.State.DEAD, -1);
}
threads.remove(streamThread);
final long cacheSizePerThread =
getCacheSizePerThread(threads.size());
resizeThreadCache(cacheSizePerThread);
+ final Collection<MemberToRemove> membersToRemove =
Collections.singletonList(new
MemberToRemove(streamThread.getGroupInstanceID()));
+
adminClient.removeMembersFromConsumerGroup(config.getString(StreamsConfig.APPLICATION_ID_CONFIG),
new RemoveMembersFromConsumerGroupOptions(membersToRemove));
+ return Optional.of(streamThread.getName());
+ }
+ }
+ }
+ log.warn("There are no threads eligible for removal");
+ } else {
+ log.warn("Cannot remove a stream thread when Kafka Streams client
is 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}.
+ *
+ * @param timeout The the length of time to wait for the thread to shutdown
+ * @throws TimeoutException if the thread does not stop in time
+ * @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(final Duration timeout) throws
TimeoutException {
+ final String msgPrefix = prepareMillisCheckFailMsgPrefix(timeout,
"timeout");
+ final long timeoutMs = validateMillisecondDuration(timeout, msgPrefix);
+ if (isRunningOrRebalancing()) {
+ synchronized (changeThreadCount) {
+ // make a copy of threads to avoid holding lock
+ for (final StreamThread streamThread : new
ArrayList<>(threads)) {
+ if (streamThread.isAlive() &&
(!streamThread.getName().equals(Thread.currentThread().getName())
+ || threads.size() == 1)) {
+ streamThread.shutdown();
+ if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
+ if
(!streamThread.waitOnThreadState(StreamThread.State.DEAD, timeoutMs)) {
+ log.warn("Thread " + streamThread.getName() +
" did not stop in the allotted time");
+ throw new TimeoutException("Thread " +
streamThread.getName() + " did not stop in the allotted time");
+ }
+ }
+ threads.remove(streamThread);
+ final long cacheSizePerThread =
getCacheSizePerThread(threads.size());
+ resizeThreadCache(cacheSizePerThread);
+ Collection<MemberToRemove> membersToRemove =
Collections.singletonList(new
MemberToRemove(streamThread.getGroupInstanceID()));
+
adminClient.removeMembersFromConsumerGroup(config.getString(StreamsConfig.APPLICATION_ID_CONFIG),
new RemoveMembersFromConsumerGroupOptions(membersToRemove));
Review comment:
Ok, this is going to be a little
tricky...`removeMembersFromConsumerGroup` is async so we have two options. (1)
just ignore the returned result and hope that it succeeded, or (2) check the
returned `KafkaFuture` and wait/make sure that it succeeded.
Probably we should go with (2) and just apply the remaining time of the
timeout. If you haven't mucked around with the KafkaFuture class before, I
believe `KafkaFuture#get(long timeout, TimeUnit unit)` is what you'd need here
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1005,11 +1008,60 @@ private StreamThread createAndAddStreamThread(final
long cacheSizePerThread, fin
|| threads.size() == 1)) {
streamThread.shutdown();
if
(!streamThread.getName().equals(Thread.currentThread().getName())) {
-
streamThread.waitOnThreadState(StreamThread.State.DEAD);
+
streamThread.waitOnThreadState(StreamThread.State.DEAD, -1);
Review comment:
To be consistent with the semantics of `KafkaStreams#close`, the
overload with no parameter should probably default to be fully blocking, ie
with a timeout of `Long.MAX_VALUE`. Also, to avoid duplicate code, I would just
have this method call `removeStreamThread(final Duration timeout)` instead of
doing everything twice. Again, something like what we do for `#close`
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -610,17 +610,32 @@ public void setStreamsUncaughtExceptionHandler(final
java.util.function.Consumer
this.streamsUncaughtExceptionHandler = streamsUncaughtExceptionHandler;
}
- public void waitOnThreadState(final StreamThread.State targetState) {
+ public boolean waitOnThreadState(final StreamThread.State targetState,
long timeoutMs) {
+ if (timeoutMs < 0) {
Review comment:
I think if you fix the semantics of `removeStreamThread()` to match that
of `close()` then there's no need for a `-1` sentinel, in which case we should
just throw an `IllegalArgumentException` here (or it's probably better to check
and throw that in the actual `removeStreamThread(timeout)` call to fail fast
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -610,17 +610,32 @@ public void setStreamsUncaughtExceptionHandler(final
java.util.function.Consumer
this.streamsUncaughtExceptionHandler = streamsUncaughtExceptionHandler;
}
- public void waitOnThreadState(final StreamThread.State targetState) {
+ public boolean waitOnThreadState(final StreamThread.State targetState,
long timeoutMs) {
+ if (timeoutMs < 0) {
+ timeoutMs = 0;
+ } else if (timeoutMs == 0) {
+ timeoutMs = Long.MAX_VALUE;
Review comment:
We definitely shouldn't modify the passed in timeout like this -- a user
should be able to pass in `0` to mean "don't block at all". Mysteriously
blocking forever when they do so would be pretty weird
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -91,6 +93,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
Review comment:
(Tbh that drives me crazy, I once spent like 4 hours debugging something
only to realize that I wasn't using the correct TimeoutException 😠)
----------------------------------------------------------------
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
