ableegoldman commented on a change in pull request #9984:
URL: https://github.com/apache/kafka/pull/9984#discussion_r565720574



##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -997,19 +1002,63 @@ private StreamThread createAndAddStreamThread(final long 
cacheSizePerThread, fin
      *         no stream threads are alive
      */
     public Optional<String> removeStreamThread() {
+        return removeStreamThread(Long.MAX_VALUE);
+    }
+
+    /**
+     * 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 {

Review comment:
       We generally don't explicitly make this part of the API, and just inform 
users through the javadocs as you've done
   ```suggestion
       public Optional<String> removeStreamThread(final Duration timeout) {
   ```

##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -997,19 +1002,63 @@ private StreamThread createAndAddStreamThread(final long 
cacheSizePerThread, fin
      *         no stream threads are alive
      */
     public Optional<String> removeStreamThread() {
+        return removeStreamThread(Long.MAX_VALUE);
+    }
+
+    /**
+     * 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

Review comment:
       ```suggestion
        * @throws org.apache.kafka.common.errors.TimeoutException if the thread 
does not stop in time
   ```

##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -88,9 +91,11 @@
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.UUID;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import org.apache.kafka.common.errors.TimeoutException;

Review comment:
       nit: move the import to the other `o.a.k.*` imports

##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1005,11 +1036,28 @@ private StreamThread createAndAddStreamThread(final 
long cacheSizePerThread, fin
                             || threads.size() == 1)) {
                         streamThread.shutdown();
                         if 
(!streamThread.getName().equals(Thread.currentThread().getName())) {
-                            
streamThread.waitOnThreadState(StreamThread.State.DEAD);
+                            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);
+                        if (streamThread.getGroupInstanceID().isPresent()) {
+                            final MemberToRemove memberToRemove = new 
MemberToRemove(streamThread.getGroupInstanceID().get());
+                            final Collection<MemberToRemove> membersToRemove = 
Collections.singletonList(memberToRemove);
+                            final RemoveMembersFromConsumerGroupResult 
removeMembersFromConsumerGroupResult = 
adminClient.removeMembersFromConsumerGroup(config.getString(StreamsConfig.APPLICATION_ID_CONFIG),
 new RemoveMembersFromConsumerGroupOptions(membersToRemove));
+                            try {
+                                
removeMembersFromConsumerGroupResult.memberResult(memberToRemove).get(timeoutMs 
- begin, TimeUnit.MILLISECONDS);
+                            } catch (final 
java.util.concurrent.TimeoutException e) {

Review comment:
       Ugh I forgot that `KafkaFuture` still throws a regular java 
TimeoutException. Such a mess -- btw we should log an error here (or a warn?) 
before rethrowing




----------------------------------------------------------------
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


Reply via email to