wcarlson5 commented on a change in pull request #11609:
URL: https://github.com/apache/kafka/pull/11609#discussion_r790055369
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -1140,6 +1140,11 @@ private void completeShutdown(final boolean cleanRun) {
topologyMetadata.unregisterThread(threadMetadata.threadName());
+ try {
+ topologyMetadata.unregisterThread(threadMetadata.threadName());
Review comment:
You might need to change line 1141?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -138,6 +138,7 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
+
maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
Review comment:
Do we really need this here?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -224,6 +225,7 @@ public void maybeWaitForNonEmptyTopology(final
Supplier<StreamThread.State> thre
} finally {
unlock();
}
+ log.info("Removed NamedTopology {} and updated topology version to
{}", topologyName, version.topologyVersion.get());
Review comment:
line 220 has the same log
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
