You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/03/07 09:15:00 UTC
[GitHub] [kafka] ableegoldman opened a new pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
ableegoldman opened a new pull request #11857:
URL: https://github.com/apache/kafka/pull/11857
We used to call `TopologyMetadata#maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion` when a thread was being unregistered/shutting down, to check if any of the futures listening for topology updates had been waiting on this thread and could be completed. Prior to invoking this we make sure to remove the current thread from the TopologyMetadata's `threadVersions` map, but this thread is actually then re-added in the `#maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion` call.
To fix this, we should break up this method into separate calls for each of its two distinct functions, updating the version and checking for topology update completion. When unregistering a thread, we should only invoke the latter method
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820595568
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -1129,13 +1129,25 @@ public void updateTaskEndMetadata(final TopicPartition topicPartition, final Lon
* added NamedTopology and create them if so, then close any tasks whose named topology no longer exists
*/
void handleTopologyUpdates() {
- tasks.maybeCreateTasksFromNewTopologies();
+ final Set<String> currentNamedTopologies = topologyMetadata.updateThreadTopologyVersion(Thread.currentThread().getName());
Review comment:
This isn't the main fix, but we were playing a little fast and loose with the topology version we were reporting having ack'ed -- tightened this up by first atomically updating the topology version and saving the set of current named topologies, then doing the actual update handling, and _then_ checking the listeners and completing any finished add/remove topology requests
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820600321
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
+ return namedTopologiesView();
+ } finally {
+ version.topologyLock.unlock();
+ }
+ }
+
+ public void maybeNotifyTopologyVersionListeners() {
+ try {
+ lock();
+ final long minThreadVersion = getMinimumThreadVersion();
+ final Iterator<TopologyVersionListener> iterator = version.activeTopologyUpdateListeners.listIterator();
+ TopologyVersionListener topologyVersionListener;
while (iterator.hasNext()) {
- topologyVersionWaiters = iterator.next();
- final long topologyVersionWaitersVersion = topologyVersionWaiters.topologyVersion;
- if (topologyVersionWaitersVersion <= threadVersions.get(threadName)) {
- if (threadVersions.values().stream().allMatch(t -> t >= topologyVersionWaitersVersion)) {
- topologyVersionWaiters.future.complete(null);
- iterator.remove();
- log.info("All threads are now on topology version {}", topologyVersionWaiters.topologyVersion);
- }
+ topologyVersionListener = iterator.next();
+ final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+ if (minThreadVersion >= topologyVersionWaitersVersion) {
Review comment:
I also refactored this slightly to optimize/clean up this method. It's less about the optimization as we should generally not have too many threads per KafkaStreams runtime, but I found it much easier to follow the logic by computing the minimum version across all threads and then completing all futures listening for the topology to be updated up to that version
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r821000029
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
+ return namedTopologiesView();
+ } finally {
+ version.topologyLock.unlock();
+ }
+ }
+
+ public void maybeNotifyTopologyVersionListeners() {
+ try {
+ lock();
+ final long minThreadVersion = getMinimumThreadVersion();
+ final Iterator<TopologyVersionListener> iterator = version.activeTopologyUpdateListeners.listIterator();
+ TopologyVersionListener topologyVersionListener;
while (iterator.hasNext()) {
- topologyVersionWaiters = iterator.next();
- final long topologyVersionWaitersVersion = topologyVersionWaiters.topologyVersion;
- if (topologyVersionWaitersVersion <= threadVersions.get(threadName)) {
- if (threadVersions.values().stream().allMatch(t -> t >= topologyVersionWaitersVersion)) {
- topologyVersionWaiters.future.complete(null);
- iterator.remove();
- log.info("All threads are now on topology version {}", topologyVersionWaiters.topologyVersion);
- }
+ topologyVersionListener = iterator.next();
+ final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+ if (minThreadVersion >= topologyVersionWaitersVersion) {
Review comment:
Ah @guozhangwang yeah the `getMinimumThreadVersion` should take care of it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r821096763
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -84,14 +84,14 @@
public AtomicLong topologyVersion = new AtomicLong(0L); // the local topology version
public ReentrantLock topologyLock = new ReentrantLock();
public Condition topologyCV = topologyLock.newCondition();
- public List<TopologyVersionWaiters> activeTopologyWaiters = new LinkedList<>();
+ public List<TopologyVersionListener> activeTopologyUpdateListeners = new LinkedList<>();
Review comment:
Good find, yeah I believe it no longer needs to be an AtomicLong, I'll change back to `long`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#issuecomment-1061503765
All test failures are unrelated, going to merge this now
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman merged pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman merged pull request #11857:
URL: https://github.com/apache/kafka/pull/11857
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820972381
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
+ return namedTopologiesView();
+ } finally {
+ version.topologyLock.unlock();
+ }
+ }
+
+ public void maybeNotifyTopologyVersionListeners() {
+ try {
+ lock();
+ final long minThreadVersion = getMinimumThreadVersion();
+ final Iterator<TopologyVersionListener> iterator = version.activeTopologyUpdateListeners.listIterator();
+ TopologyVersionListener topologyVersionListener;
while (iterator.hasNext()) {
- topologyVersionWaiters = iterator.next();
- final long topologyVersionWaitersVersion = topologyVersionWaiters.topologyVersion;
- if (topologyVersionWaitersVersion <= threadVersions.get(threadName)) {
- if (threadVersions.values().stream().allMatch(t -> t >= topologyVersionWaitersVersion)) {
- topologyVersionWaiters.future.complete(null);
- iterator.remove();
- log.info("All threads are now on topology version {}", topologyVersionWaiters.topologyVersion);
- }
+ topologyVersionListener = iterator.next();
+ final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+ if (minThreadVersion >= topologyVersionWaitersVersion) {
Review comment:
The listeners are for the caller threads not stream threads right? I thought since the thread is removed, it would not be counted in the `getMinimumThreadVersion()` and hence would not block the listeners to be removed.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820593601
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -881,17 +881,15 @@ private void initializeAndRestorePhase() {
// Check if the topology has been updated since we last checked, ie via #addNamedTopology or #removeNamedTopology
private void checkForTopologyUpdates() {
if (topologyMetadata.isEmpty() || topologyMetadata.needsUpdate(getName())) {
+ log.info("StreamThread has detected an update to the topology");
+
taskManager.handleTopologyUpdates();
- log.info("StreamThread has detected an update to the topology, triggering a rebalance to refresh the assignment");
- if (topologyMetadata.isEmpty()) {
- mainConsumer.unsubscribe();
- }
- topologyMetadata.maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(getName());
Review comment:
All of this has been moved into `taskManager#handleTopologyUpdates`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#issuecomment-1060371876
cc @wcarlson5
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820993077
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
+ return namedTopologiesView();
+ } finally {
+ version.topologyLock.unlock();
+ }
+ }
+
+ public void maybeNotifyTopologyVersionListeners() {
+ try {
+ lock();
+ final long minThreadVersion = getMinimumThreadVersion();
+ final Iterator<TopologyVersionListener> iterator = version.activeTopologyUpdateListeners.listIterator();
+ TopologyVersionListener topologyVersionListener;
while (iterator.hasNext()) {
- topologyVersionWaiters = iterator.next();
- final long topologyVersionWaitersVersion = topologyVersionWaiters.topologyVersion;
- if (topologyVersionWaitersVersion <= threadVersions.get(threadName)) {
- if (threadVersions.values().stream().allMatch(t -> t >= topologyVersionWaitersVersion)) {
- topologyVersionWaiters.future.complete(null);
- iterator.remove();
- log.info("All threads are now on topology version {}", topologyVersionWaiters.topologyVersion);
- }
+ topologyVersionListener = iterator.next();
+ final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+ if (minThreadVersion >= topologyVersionWaitersVersion) {
Review comment:
Hmm... just to make sure we are talking about `version.activeTopologyUpdateListeners` right? These listeners are for the calling thread of the `removeNamedTopology / addNamedTopology / start`, which would get the wraped futures these listeners are constructed on.
Anyways, my understanding is that when a thread is removed, the `getMinimumThreadVersion` returned version would not take that removed thread into consideration, so that even the removed thread's version is low it would not block the future being completed.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
+ return namedTopologiesView();
+ } finally {
+ version.topologyLock.unlock();
+ }
+ }
+
+ public void maybeNotifyTopologyVersionListeners() {
+ try {
+ lock();
+ final long minThreadVersion = getMinimumThreadVersion();
+ final Iterator<TopologyVersionListener> iterator = version.activeTopologyUpdateListeners.listIterator();
+ TopologyVersionListener topologyVersionListener;
Review comment:
nit: this `TopologyVersionListener topologyVersionListener` could be declared within the while loop.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820981114
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
+ return namedTopologiesView();
+ } finally {
+ version.topologyLock.unlock();
+ }
+ }
+
+ public void maybeNotifyTopologyVersionListeners() {
+ try {
+ lock();
+ final long minThreadVersion = getMinimumThreadVersion();
+ final Iterator<TopologyVersionListener> iterator = version.activeTopologyUpdateListeners.listIterator();
+ TopologyVersionListener topologyVersionListener;
while (iterator.hasNext()) {
- topologyVersionWaiters = iterator.next();
- final long topologyVersionWaitersVersion = topologyVersionWaiters.topologyVersion;
- if (topologyVersionWaitersVersion <= threadVersions.get(threadName)) {
- if (threadVersions.values().stream().allMatch(t -> t >= topologyVersionWaitersVersion)) {
- topologyVersionWaiters.future.complete(null);
- iterator.remove();
- log.info("All threads are now on topology version {}", topologyVersionWaiters.topologyVersion);
- }
+ topologyVersionListener = iterator.next();
+ final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+ if (minThreadVersion >= topologyVersionWaitersVersion) {
Review comment:
No, the listeners are for stream threads. They get added in the task manager. Once all threads are at the version the future blocking the calling thread is completed.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r821006143
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -84,14 +84,14 @@
public AtomicLong topologyVersion = new AtomicLong(0L); // the local topology version
public ReentrantLock topologyLock = new ReentrantLock();
public Condition topologyCV = topologyLock.newCondition();
- public List<TopologyVersionWaiters> activeTopologyWaiters = new LinkedList<>();
+ public List<TopologyVersionListener> activeTopologyUpdateListeners = new LinkedList<>();
Review comment:
Also another quick question regarding why we need to keep `topologyVersion` an atomic long? Seems besides the getters all of its updators are under the lock as well.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r821099068
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -84,14 +84,14 @@
public AtomicLong topologyVersion = new AtomicLong(0L); // the local topology version
public ReentrantLock topologyLock = new ReentrantLock();
public Condition topologyCV = topologyLock.newCondition();
- public List<TopologyVersionWaiters> activeTopologyWaiters = new LinkedList<>();
+ public List<TopologyVersionListener> activeTopologyUpdateListeners = new LinkedList<>();
Review comment:
Oh right actually no, we. do still need it to be an AtomicLong as we check it in the StreamThread main loop when looking for topology updates. And obviously we don't want to have to grab the full lock for that
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820926850
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
+ return namedTopologiesView();
+ } finally {
+ version.topologyLock.unlock();
+ }
+ }
+
+ public void maybeNotifyTopologyVersionListeners() {
+ try {
+ lock();
+ final long minThreadVersion = getMinimumThreadVersion();
+ final Iterator<TopologyVersionListener> iterator = version.activeTopologyUpdateListeners.listIterator();
+ TopologyVersionListener topologyVersionListener;
while (iterator.hasNext()) {
- topologyVersionWaiters = iterator.next();
- final long topologyVersionWaitersVersion = topologyVersionWaiters.topologyVersion;
- if (topologyVersionWaitersVersion <= threadVersions.get(threadName)) {
- if (threadVersions.values().stream().allMatch(t -> t >= topologyVersionWaitersVersion)) {
- topologyVersionWaiters.future.complete(null);
- iterator.remove();
- log.info("All threads are now on topology version {}", topologyVersionWaiters.topologyVersion);
- }
+ topologyVersionListener = iterator.next();
+ final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+ if (minThreadVersion >= topologyVersionWaitersVersion) {
Review comment:
I think we also want to remove the listeners for threads that were removed as well right?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r821100612
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
+ return namedTopologiesView();
+ } finally {
+ version.topologyLock.unlock();
+ }
+ }
+
+ public void maybeNotifyTopologyVersionListeners() {
+ try {
+ lock();
+ final long minThreadVersion = getMinimumThreadVersion();
+ final Iterator<TopologyVersionListener> iterator = version.activeTopologyUpdateListeners.listIterator();
+ TopologyVersionListener topologyVersionListener;
while (iterator.hasNext()) {
- topologyVersionWaiters = iterator.next();
- final long topologyVersionWaitersVersion = topologyVersionWaiters.topologyVersion;
- if (topologyVersionWaitersVersion <= threadVersions.get(threadName)) {
- if (threadVersions.values().stream().allMatch(t -> t >= topologyVersionWaitersVersion)) {
- topologyVersionWaiters.future.complete(null);
- iterator.remove();
- log.info("All threads are now on topology version {}", topologyVersionWaiters.topologyVersion);
- }
+ topologyVersionListener = iterator.next();
+ final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+ if (minThreadVersion >= topologyVersionWaitersVersion) {
Review comment:
Yeah I think this was already resolved but just to clarify for anyone else reading this/ourselves in the future, yes, the listeners are for the callers of add/removeNamedTopology 👍
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820598589
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -161,35 +161,47 @@ public void registerThread(final String threadName) {
public void unregisterThread(final String threadName) {
threadVersions.remove(threadName);
- maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(threadName);
+ maybeNotifyTopologyVersionListeners();
}
public TaskExecutionMetadata taskExecutionMetadata() {
return taskExecutionMetadata;
}
- public void maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(final String threadName) {
+ public Set<String> updateThreadTopologyVersion(final String threadName) {
try {
- lock();
- final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
- TopologyVersionWaiters topologyVersionWaiters;
+ version.topologyLock.lock();
threadVersions.put(threadName, topologyVersion());
Review comment:
@wcarlson5 / @guozhangwang / @vvcephei This is the main fix -- we need to split out the version update where we add the current thread with the latest topology version to this `threadVersions` map, since this of course should only be done when we're reacting to a topology update.
The other function of the method was to check whether we could complete any of the queued listeners, which is why we were invoking this when shutting down a thread. Splitting this out into a separate method avoids ghost threads being left behind in the `threadVersions` map
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11857: KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11857:
URL: https://github.com/apache/kafka/pull/11857#discussion_r820595967
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java
##########
@@ -84,14 +84,14 @@
public AtomicLong topologyVersion = new AtomicLong(0L); // the local topology version
public ReentrantLock topologyLock = new ReentrantLock();
public Condition topologyCV = topologyLock.newCondition();
- public List<TopologyVersionWaiters> activeTopologyWaiters = new LinkedList<>();
+ public List<TopologyVersionListener> activeTopologyUpdateListeners = new LinkedList<>();
Review comment:
Just renamed from `waiters` to `listeners`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org