Skip to content

[SPARK-23416][SS] Add a specific stop method for ContinuousExecution. #21384

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
wants to merge 1 commit into from
Closed

[SPARK-23416][SS] Add a specific stop method for ContinuousExecution. #21384

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
18 changes: 18 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,24 @@ class MicroBatchExecution(
_logicalPlan
}

/**
* Signals to the thread executing micro-batches that it should stop running after the next
* batch. This method blocks until the thread stops running.
*/
override def stop(): Unit = {
// Set the state to TERMINATED so that the batching thread knows that it was interrupted
// intentionally
state.set(TERMINATED)
if (queryExecutionThread.isAlive) {
sparkSession.sparkContext.cancelJobGroup(runId.toString)
queryExecutionThread.interrupt()
queryExecutionThread.join()
// microBatchThread may spawn new jobs, so we need to cancel again to prevent a leak
sparkSession.sparkContext.cancelJobGroup(runId.toString)
}
logInfo(s"Query $prettyIdString was stopped")
}

/**
* Repeatedly attempts to run batches as data arrives.
*/
Expand Down
18 changes: 0 additions & 18 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -378,24 +378,6 @@ abstract class StreamExecution(
}
}

/**
* Signals to the thread executing micro-batches that it should stop running after the next
* batch. This method blocks until the thread stops running.
*/
override def stop(): Unit = {
// Set the state to TERMINATED so that the batching thread knows that it was interrupted
// intentionally
state.set(TERMINATED)
if (queryExecutionThread.isAlive) {
sparkSession.sparkContext.cancelJobGroup(runId.toString)
queryExecutionThread.interrupt()
queryExecutionThread.join()
// microBatchThread may spawn new jobs, so we need to cancel again to prevent a leak
sparkSession.sparkContext.cancelJobGroup(runId.toString)
}
logInfo(s"Query $prettyIdString was stopped")
}

/**
* Blocks the current thread until processing for data from the given `source` has reached at
* least the given `Offset`. This method is intended for use primarily when writing tests.
Expand Down
16 changes: 16 additions & 0 deletions .../main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,6 +356,22 @@ class ContinuousExecution(
}
}
}

/**
* Stops the query execution thread to terminate the query.
*/
override def stop(): Unit = {
// Set the state to TERMINATED so that the batching thread knows that it was interrupted
// intentionally
state.set(TERMINATED)
if (queryExecutionThread.isAlive) {
// The query execution thread will clean itself up in the finally clause of runContinuous.
// We just need to interrupt the long running job.
queryExecutionThread.interrupt()
queryExecutionThread.join()
Copy link
Member

@dongjoon-hyun dongjoon-hyun May 22, 2018

Choose a reason for hiding this comment

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

Thank you for pinging me, @jose-torres .
So, technically, two sparkSession.sparkContext.cancelJobGroup(runId.toString) are removed in continuousExecution?

Copy link
Contributor Author

@jose-torres jose-torres May 22, 2018

Choose a reason for hiding this comment

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

Correct. The remaining one in the finally clause of runContinuous() is sufficient, because jobs are only started within that method.

}
logInfo(s"Query $prettyIdString was stopped")
}
}

object ContinuousExecution {
Expand Down