Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/16186#discussion_r91419859
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala
---
@@ -35,12 +43,24 @@ class StreamingQueryListenerBus(sparkListenerBus:
LiveListenerBus)
sparkListenerBus.addListener(this)
/**
- * Post a StreamingQueryListener event to the Spark listener bus
asynchronously. This event will
- * be dispatched to all StreamingQueryListener in the thread of the
Spark listener bus.
+ * RunIds of active queries whose events are supposed to be forwarded by
this ListenerBus
+ * to registered `StreamingQueryListeners`. Note that this list needs to
be maintained separately
+ * from the `StreamingQueryManager.activeQueries` because a terminated
query is cleared from
+ * `StreamingQueryManager.activeQueries` as soon as it is stopped, but
the this ListenerBus must
+ * clear a query only after the termination event of that query has been
posted.
+ */
+ private val activeQueryRunIds = new mutable.HashSet[UUID]
--- End diff --
Even if this behavior was not different (that is, all async), this
component should not be responsible preventing concurrent runs. This component
should be simple and not deal with such issues. I have added more docs
regarding why runIds instead of ids
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
