ableegoldman commented on a change in pull request #9984: URL: https://github.com/apache/kafka/pull/9984#discussion_r565731661
########## File path: streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java ########## @@ -319,9 +319,9 @@ private void prepareStreamThread(final StreamThread thread, final boolean termin StreamThread.State.PARTITIONS_ASSIGNED); return null; }).anyTimes(); + EasyMock.expect(thread.getGroupInstanceID()).andReturn(Optional.empty()).anyTimes(); Review comment: ```suggestion EasyMock.expect(thread.getGroupInstanceID()).andStubReturn(Optional.empty()); ``` ########## File path: streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java ########## @@ -180,6 +182,19 @@ public void shouldRemoveStreamThread() throws Exception { } } + @Test + public void shouldnNotRemoveStreamThreadWithTimeout() throws Exception { + try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) { + addStreamStateChangeListener(kafkaStreams); + startStreamsAndWaitForRunning(kafkaStreams); + + final int oldThreadCount = kafkaStreams.localThreadsMetadata().size(); + stateTransitionHistory.clear(); + assertThrows(TimeoutException.class, () -> kafkaStreams.removeStreamThread(Duration.ZERO.minus(DEFAULT_DURATION))); Review comment: It's a bit weird to test this by passing in a negative timeout but I don't have any good ideas for forcing it to exceed the timeout 😕 ########## File path: streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java ########## @@ -180,6 +182,19 @@ public void shouldRemoveStreamThread() throws Exception { } } + @Test + public void shouldnNotRemoveStreamThreadWithTimeout() throws Exception { Review comment: ```suggestion public void shouldNotRemoveStreamThreadWithinTimeout() throws Exception { ``` ########## 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 { + final String msgPrefix = prepareMillisCheckFailMsgPrefix(timeout, "timeout"); + final long timeoutMs = validateMillisecondDuration(timeout, msgPrefix); + return removeStreamThread(timeoutMs); + } + + private Optional<String> removeStreamThread(final long timeoutMs) throws TimeoutException { + final long begin = time.milliseconds(); 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)) { + final Optional<String> groupInstanceID = streamThread.getGroupInstanceID(); 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"); Review comment: Hm actually now that I think about it, we should probably continue with the cleanup to leave the app in a good state even if the shutdown is taking a long time. eg just store the exception for now and then rethrow it at the end of this method -- WDYT? ---------------------------------------------------------------- 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