Skip to content

[SPARK-29352][SQL][SS] Track active streaming queries in the SparkSession.sharedState #26018

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
brkyvz wants to merge 8 commits into from
Closed

[SPARK-29352][SQL][SS] Track active streaming queries in the SparkSession.sharedState #26018

Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
08be288
Track active streaming queries in the SparkSession.sharedState
brkyvz Oct 3, 2019
b95474a
Update SharedState.scala
brkyvz Oct 4, 2019
b5eb762
Add tests
brkyvz Oct 4, 2019
a91f017
Merge branch 'sharedStreamingQueryManager' of github.com:brkyvz/spark…
brkyvz Oct 4, 2019
0ddb7e5
add docs
brkyvz Oct 4, 2019
b808d33
Update StreamingQueryManager.scala
brkyvz Oct 10, 2019
d275ec6
Update StreamingQueryManager.scala
brkyvz Oct 13, 2019
37b43ec
Update StreamingQueryManagerSuite.scala
brkyvz Oct 22, 2019
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
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -355,7 +355,7 @@ class StreamingQueryManager private[sql] (sparkSession: SparkSession) extends Lo
// Make sure no other query with same id is active across all sessions
val activeOption =
Option(sparkSession.sharedState.activeStreamingQueries.putIfAbsent(query.id, this))
if (activeOption.isDefined) {
if (activeOption.isDefined || activeQueries.values.exists(_.id == query.id) {
Copy link
Contributor

@HeartSaVioR HeartSaVioR Oct 11, 2019

Choose a reason for hiding this comment

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

Just curious: are you seeing the actual case where activeOption is None but activeQueries contain such query? I'm not seeing the case, though I don't think adding this would hurt.

Copy link
Member

@dongjoon-hyun dongjoon-hyun Oct 13, 2019 •
edited
Loading

Choose a reason for hiding this comment

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

It seems that one ) is missed after query.id). Style check and compilation failure.

[error] /home/jenkins/workspace/SparkPullRequestBuilder@2/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala: Expected token RPAREN but got Token(DOT,.,14050,.)
[error] Total time: 20 s, completed Oct 12, 2019 4:11:49 PM

throw new IllegalStateException(
s"Cannot start query with id ${query.id} as another query with same id is " +
s"already active. Perhaps you are attempting to restart a query from checkpoint " +
Expand Down