Skip to content

[SPARK-18144][SQL] logging StreamingQueryListener$QueryStartedEvent #15675

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
CodingCat wants to merge 8 commits into from
Closed

[SPARK-18144][SQL] logging StreamingQueryListener$QueryStartedEvent #15675

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
5a8dc87
improve the doc for "spark.memory.offHeap.size"
CodingCat Mar 7, 2016
46788bd
fix
CodingCat Mar 7, 2016
766c67d
postToAll has to be called in the same thread
CodingCat Oct 28, 2016
24bcd7b
fix the misunderstanding of the problem
CodingCat Oct 29, 2016
2c2147e
guarantee thread-safety
CodingCat Oct 31, 2016
8d2b3a6
fix
CodingCat Oct 31, 2016
4b29d8f
de-duplicate QueryStartedEvent and add test
CodingCat Nov 1, 2016
00dd3ce
address shixiong's comments
CodingCat Nov 2, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion ...e/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@ class StreamingQueryListenerBus(sparkListenerBus: LiveListenerBus)
def post(event: StreamingQueryListener.Event) {
event match {
Copy link
Contributor Author

@CodingCat CodingCat Oct 28, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

actually I didn't see any reason to distinguish these two types of event when posting

case s: QueryStartedEvent =>
sparkListenerBus.post(s)
Copy link
Member

@zsxwing zsxwing Oct 31, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

QueryStartedEvent will be sent twice to StreamingQueryListener. There is a thread local variable in LiveListenerBus. You can use it to ignore duplicated QueryStartedEvent if it's posted in the listener thread.

Copy link
Member

@zsxwing zsxwing Oct 31, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is hacky, it's better to have some tests to make sure we won't break things in future when removing the hacky codes.

// post to local listeners to trigger callbacks
postToAll(s)
case _ =>
sparkListenerBus.post(event)
Expand All @@ -50,7 +52,13 @@ class StreamingQueryListenerBus(sparkListenerBus: LiveListenerBus)
override def onOtherEvent(event: SparkListenerEvent): Unit = {
event match {
case e: StreamingQueryListener.Event =>
postToAll(e)
// SPARK-18144: we broadcast QueryStartedEvent to all listeners attached to this bus
// synchronously and the ones attached to LiveListenerBus asynchronously. Therefore,
// we need to ignore QueryStartedEvent if this method is called within SparkListenerBus
// thread
if (!LiveListenerBus.withinListenerThread.value || !e.isInstanceOf[QueryStartedEvent]) {
postToAll(e)
}
case _ =>
}
}
Expand Down
7 changes: 6 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,7 +290,10 @@ class StreamingQuerySuite extends StreamTest with BeforeAndAfter with Logging {
// A StreamingQueryListener that gets the query status after the first completed trigger
val listener = new StreamingQueryListener {
@volatile var firstStatus: StreamingQueryStatus = null
override def onQueryStarted(queryStarted: QueryStartedEvent): Unit = { }
@volatile var queryStartedEvent = 0
Copy link
Member

@zsxwing zsxwing Nov 1, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: please add @volatile

override def onQueryStarted(queryStarted: QueryStartedEvent): Unit = {
queryStartedEvent += 1
}
override def onQueryProgress(queryProgress: QueryProgressEvent): Unit = {
if (firstStatus == null) firstStatus = queryProgress.queryStatus
}
Expand All @@ -303,6 +306,8 @@ class StreamingQuerySuite extends StreamTest with BeforeAndAfter with Logging {
q.processAllAvailable()
eventually(timeout(streamingTimeout)) {
assert(listener.firstStatus != null)
// test if QueryStartedEvent callback is called for only once
assert(listener.queryStartedEvent === 1)
}
listener.firstStatus
} finally {
Expand Down