[SPARK-19617][SS]Fix the race condition when starting and stopping a query quickly #16947
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,9 +17,9 @@ | |
|
|
||
| package org.apache.spark.sql.execution.streaming | ||
|
|
||
| import java.util.UUID | ||
| import java.util.concurrent.{CountDownLatch, TimeUnit} | ||
| import java.util.concurrent.atomic.AtomicReference | ||
| import java.util.concurrent.locks.ReentrantLock | ||
|
|
||
| import scala.collection.mutable.ArrayBuffer | ||
|
|
@@ -158,8 +158,7 @@ class StreamExecution( | |
| } | ||
|
|
||
| /** Defines the internal state of execution */ | ||
| private val state = new AtomicReference[State](INITIALIZING) | ||
|
|
||
| @volatile | ||
| var lastExecution: QueryExecution = _ | ||
|
|
@@ -179,8 +178,8 @@ class StreamExecution( | |
|
|
||
| /** | ||
| * The thread that runs the micro-batches of this stream. Note that this thread must be | ||
| * [[org.apache.spark.util.UninterruptibleThread]] to workaround KAFKA-1894: interrupting a | ||
| * running `KafkaConsumer` may cause endless loop. | ||
| */ | ||
| val microBatchThread = | ||
| new StreamExecutionThread(s"stream execution thread for $prettyIdString") { | ||
|
|
@@ -201,10 +200,10 @@ class StreamExecution( | |
| val offsetLog = new OffsetSeqLog(sparkSession, checkpointFile("offsets")) | ||
|
|
||
| /** Whether all fields of the query have been initialized */ | ||
| private def isInitialized: Boolean = state.get != INITIALIZING | ||
|
|
||
| /** Whether the query is currently active or not */ | ||
| override def isActive: Boolean = state.get != TERMINATED | ||
|
|
||
| /** Returns the [[StreamingQueryException]] if the query was terminated by an exception. */ | ||
| override def exception: Option[StreamingQueryException] = Option(streamDeathCause) | ||
|
|
@@ -250,53 +249,56 @@ class StreamExecution( | |
| updateStatusMessage("Initializing sources") | ||
| // force initialization of the logical plan so that the sources can be created | ||
| logicalPlan | ||
| if (state.compareAndSet(INITIALIZING, ACTIVE)) { | ||
|
Member
Author
There was a problem hiding this comment. Most changes here are space changes. You can use https://github.com/apache/spark/pull/16947/files?w=1 to review it. |
||
| // Unblock `awaitInitialization` | ||
| initializationLatch.countDown() | ||
|
|
||
| triggerExecutor.execute(() => { | ||
| startTrigger() | ||
|
|
||
| val continueToRun = | ||
| if (isActive) { | ||
| reportTimeTaken("triggerExecution") { | ||
| if (currentBatchId < 0) { | ||
| // We'll do this initialization only once | ||
| populateStartOffsets() | ||
| logDebug(s"Stream running from $committedOffsets to $availableOffsets") | ||
| } else { | ||
| constructNextBatch() | ||
| } | ||
| if (dataAvailable) { | ||
| currentStatus = currentStatus.copy(isDataAvailable = true) | ||
| updateStatusMessage("Processing new data") | ||
| runBatch() | ||
| } | ||
| } | ||
|
|
||
| // Report trigger as finished and construct progress object. | ||
| finishTrigger(dataAvailable) | ||
| if (dataAvailable) { | ||
| // We'll increase currentBatchId after we complete processing current batch's data | ||
| currentBatchId += 1 | ||
| } else { | ||
| currentStatus = currentStatus.copy(isDataAvailable = false) | ||
| updateStatusMessage("Waiting for data to arrive") | ||
| Thread.sleep(pollingDelayMs) | ||
| } | ||
| true | ||
| } else { | ||
| false | ||
| } | ||
|
|
||
| // Update committed offsets. | ||
| committedOffsets ++= availableOffsets | ||
| updateStatusMessage("Waiting for next trigger") | ||
| continueToRun | ||
| }) | ||
| updateStatusMessage("Stopped") | ||
| } else { | ||
| // `stop()` is already called. Let `finally` finish the cleanup. | ||
| } | ||
| } catch { | ||
| case _: InterruptedException if state.get == TERMINATED => // interrupted by stop() | ||
| updateStatusMessage("Stopped") | ||
| case e: Throwable => | ||
| streamDeathCause = new StreamingQueryException( | ||
|
|
@@ -319,7 +321,7 @@ class StreamExecution( | |
| initializationLatch.countDown() | ||
|
|
||
| try { | ||
| state.set(TERMINATED) | ||
| currentStatus = status.copy(isTriggerActive = false, isDataAvailable = false) | ||
|
|
||
| // Update metrics and status | ||
|
|
@@ -563,7 +565,7 @@ class StreamExecution( | |
| override def stop(): Unit = { | ||
| // Set the state to TERMINATED so that the batching thread knows that it was interrupted | ||
| // intentionally | ||
| state.set(TERMINATED) | ||
| if (microBatchThread.isAlive) { | ||
| microBatchThread.interrupt() | ||
| microBatchThread.join() | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Didnt we disable interrupt because with local files, hadoop used shell commands to do file manipulation which could hang when interrupted? Are we removing this now because that has been fixed in hadoop?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes. We dropped the support to Hadoop 2.5 and earlier versions.