wcarlson5 commented on a change in pull request #9984:
URL: https://github.com/apache/kafka/pull/9984#discussion_r565607533
##########
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:
for the non timeout uses
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -1147,6 +1162,10 @@ public String toString(final String indent) {
return indent + "\tStreamsThread threadId: " + getName() + "\n" +
taskManager.toString(indent);
}
+ public String getGroupInstanceID(){
+ return mainConsumer.groupMetadata().groupInstanceId().orElse("");
Review comment:
It seems easier to get it form here than the config. It looked like I
might have how to manipulate strings in that case
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1005,11 +1007,56 @@ 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);
+ Collection<MemberToRemove> membersToRemove =
Collections.singletonList(new
MemberToRemove(streamThread.getGroupInstanceID()));
Review comment:
I ended up getting the `group.instance.id` from the streamThread
----------------------------------------------------------------
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
