This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new acc92cd68f95 [SPARK-58131][SS][RTM] Support async progress tracking
for Real-Time Mode
acc92cd68f95 is described below
commit acc92cd68f959e7f3223bba36a797069a4f216ab
Author: Boyang Jerry Peng <[email protected]>
AuthorDate: Tue Jul 21 07:52:24 2026 +0900
[SPARK-58131][SS][RTM] Support async progress tracking for Real-Time Mode
### What changes were proposed in this pull request?
Enable Async Progress Tracking (APT) for stateless Real-Time Mode (RTM)
streaming queries. Previously the combination was explicitly rejected with
`STREAMING_REAL_TIME_MODE.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED`.
- `StreamingQueryManager`: remove the RTM+APT rejection guard; enable APT
by default for stateless RTM queries (gated by a new internal conf, with the
`asyncProgressTrackingEnabled` writer option still taking precedence).
- `SQLConf`: add
`spark.sql.streaming.realTimeMode.asyncProgressTrackingByDefault.enabled`.
- `MicroBatchExecution`: extract the inline RTM end-of-batch offset
handling into `populateBatchOffsetsForRTM` and
`signalProcessAllAvailableIfRealTimeMode` so the APT subclass can reuse it
(behavior-preserving refactor).
- `AsyncProgressTrackingMicroBatchExecution`: RTM-aware async offset/commit
log writes, with fail-fast error propagation for async write failures.
- `StreamingErrors` + `error-conditions.json`: add the log-write-failure
and RTM checkpoint-interval error conditions.
- Docs: RTM page no longer states APT is unsupported.
### Why are the changes needed?
APT lets RTM stateless queries checkpoint offset/commit logs
asynchronously, keeping checkpoint I/O off the critical path for low-latency
processing.
### Does this PR introduce _any_ user-facing change?
Yes. Stateless Real-Time Mode queries now use async progress tracking by
default, and `asyncProgressTrackingEnabled=true` is now accepted (rather than
rejected) in RTM.
### How was this patch tested?
New `AsyncProgressTrackingRealTimeModeSuite` (happy path, default-on
enablement, user-override precedence, stateful exclusion, recovery,
back-pressure, and fail-fast on both commit-log and offset-log write failures).
Existing APT and RTM suites and `SparkThrowableSuite` pass.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored-by: Claude Code (Opus 4.8)
Closes #57259 from jerrypeng/rtm-async-progress-tracking.
Authored-by: Boyang Jerry Peng <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 28 +-
docs/streaming/real-time-mode.md | 9 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 11 +
.../spark/sql/classic/StreamingQueryManager.scala | 27 +-
.../sql/execution/streaming/StreamingErrors.scala | 35 ++
.../AsyncProgressTrackingMicroBatchExecution.scala | 257 ++++++++++----
.../streaming/runtime/MicroBatchExecution.scala | 119 ++++---
.../AsyncProgressTrackingRealTimeModeSuite.scala | 381 +++++++++++++++++++++
.../FailureInjectionCheckpointFileManager.scala | 18 +-
.../sql/streaming/StreamRealTimeModeSuite.scala | 16 -
.../apache/spark/sql/streaming/StreamTest.scala | 27 +-
11 files changed, 779 insertions(+), 149 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index a72e279f25fb..eaa6471e29b9 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -7311,6 +7311,30 @@
],
"sqlState" : "42K09"
},
+ "STREAMING_ASYNC_OPERATION_FAILED" : {
+ "message" : [
+ "Asynchronous streaming operation failed with exception: <message>"
+ ],
+ "sqlState" : "XXKST"
+ },
+ "STREAMING_CHECKPOINT_LOG_WRITE_FAILURE" : {
+ "message" : [
+ "Failed to write a streaming query checkpoint log entry."
+ ],
+ "subClass" : {
+ "COMMIT_LOG" : {
+ "message" : [
+ "Failed to write commit log entry for batch <batchId> at checkpoint
location <checkpointLocation>. Check that the checkpoint location is writable
and retry the query."
+ ]
+ },
+ "OFFSET_LOG" : {
+ "message" : [
+ "Failed to write offset log entry for batch <batchId> at checkpoint
location <checkpointLocation>. Check that the checkpoint location is writable
and retry the query."
+ ]
+ }
+ },
+ "sqlState" : "58030"
+ },
"STREAMING_CHECKPOINT_MISSING_METADATA_FILE" : {
"message" : [
"Checkpoint location <checkpointLocation> is in an inconsistent state:
the metadata file is missing but offset and/or commit logs contain data. Please
restore the metadata file or create a new checkpoint directory."
@@ -7416,9 +7440,9 @@
"Streaming real-time mode has the following limitation:"
],
"subClass" : {
- "ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED" : {
+ "ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_NON_ZERO" : {
"message" : [
- "Async progress tracking is not supported in real-time mode. Set
option asyncProgressTrackingEnabled to false and retry your query."
+ "The checkpointing interval for async progress tracking must be set
to 0, which means each progress update is checkpointed. Set option
asyncProgressTrackingCheckpointIntervalMs to 0 in DataStreamWriter options and
retry your query."
]
},
"IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED" : {
diff --git a/docs/streaming/real-time-mode.md b/docs/streaming/real-time-mode.md
index 15731dfb36a7..51a8ba04c073 100644
--- a/docs/streaming/real-time-mode.md
+++ b/docs/streaming/real-time-mode.md
@@ -85,6 +85,12 @@ Choosing the batch duration is a trade-off:
The duration is set on the Real-time trigger, as shown under
[Enabling Real-time Mode](#enabling-real-time-mode).
+Progress is committed using **asynchronous progress tracking**: the offset and
commit logs are
+written off the record-processing path so that checkpointing does not stall
processing. This is
+enabled automatically for Real-time Mode queries and every batch is
checkpointed (the async
+progress tracking checkpoint interval is fixed at 0 in Real-time Mode). It can
be turned off with
+the `asyncProgressTrackingEnabled` writer option, in which case progress is
committed synchronously.
+
## Comparison with Other Modes
The table below summarizes how Real-time Mode relates to the default
micro-batch engine and to the
@@ -268,8 +274,7 @@ Stateful operations of any kind are not supported in this
release. This includes
aggregations, `dropDuplicates` / `dropDuplicatesWithinWatermark`,
stream-stream joins, `repartition`
and other operations that introduce a shuffle, and stateful operators such as
`flatMapGroupsWithState` and `transformWithState`. Support for stateful
operations is planned
-starting in Spark 4.3. Asynchronous progress tracking is also not supported;
enabling it fails with
-`STREAMING_REAL_TIME_MODE.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED`.
+starting in Spark 4.3.
## Fault Tolerance
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 0b3f4693853a..2349f3dca56f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3762,6 +3762,17 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STREAMING_ASYNC_PROGRESS_TRACKING_REAL_TIME_MODE_ENABLED_BY_DEFAULT =
buildConf(
+ "spark.sql.streaming.realTimeMode.asyncProgressTrackingByDefault.enabled")
+ .internal()
+ .doc("Whether asynchronous progress tracking should be enabled in
real-time mode by " +
+ "default for stateless queries. If explicitly set, the stream writer
option for async " +
+ "progress tracking will still take precedence over this flag.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
val VARIABLE_SUBSTITUTE_ENABLED =
buildConf("spark.sql.variable.substitute")
.doc("This enables substitution using syntax like `${var}`,
`${system:var}`, " +
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
index 2a59b8f08f9e..c934be69bc69 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala
@@ -168,9 +168,22 @@ class StreamingQueryManager private[sql] (
listenerBus.post(event)
}
- private def useAsyncProgressTracking(extraOptions: Map[String, String]):
Boolean = {
- extraOptions.getOrElse(
-
AsyncProgressTrackingMicroBatchExecution.ASYNC_PROGRESS_TRACKING_ENABLED,
"false").toBoolean
+ private def isAnalysedPlanStateful(analysedPlan: WriteToStream): Boolean = {
+ analysedPlan.exists { op => op.isStateful }
+ }
+
+ private def useAsyncProgressTracking(
+ extraOptions: Map[String, String],
+ trigger: Trigger,
+ analysedPlan: WriteToStream
+ ): Boolean = {
+
extraOptions.get(AsyncProgressTrackingMicroBatchExecution.ASYNC_PROGRESS_TRACKING_ENABLED)
+ .map(_.toBoolean)
+ // We enable APT in RTM by default whenever the query is stateless.
+ .getOrElse(trigger.isInstanceOf[RealTimeTrigger] &&
+ !isAnalysedPlanStateful(analysedPlan) &&
+ sparkSession.sessionState.conf.getConf(
+
SQLConf.STREAMING_ASYNC_PROGRESS_TRACKING_REAL_TIME_MODE_ENABLED_BY_DEFAULT))
}
// scalastyle:off argcount
@@ -242,12 +255,8 @@ class StreamingQueryManager private[sql] (
extraOptions,
analyzedStreamWritePlan))
case _ =>
- val microBatchExecution = if (useAsyncProgressTracking(extraOptions)) {
- if (trigger.isInstanceOf[RealTimeTrigger]) {
- throw new SparkIllegalArgumentException(
- errorClass =
"STREAMING_REAL_TIME_MODE.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED"
- )
- }
+ val useApt = useAsyncProgressTracking(extraOptions, trigger,
analyzedStreamWritePlan)
+ val microBatchExecution = if (useApt) {
// Sink evolution persists per-sink metadata via the V3 commit log
written in
// MicroBatchExecution.markMicroBatchEnd, which
AsyncProgressTrackingMicroBatchExecution
// overrides with an async write that only emits V1 commit metadata.
The sink metadata
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingErrors.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingErrors.scala
index 5efb5e50cd8f..d8f1498233f6 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingErrors.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingErrors.scala
@@ -22,6 +22,41 @@ import org.apache.spark.{SparkException,
SparkRuntimeException}
* Object for grouping error messages from streaming query exceptions
*/
object StreamingErrors {
+ /**
+ * Error condition and subclasses for failures writing a streaming
checkpoint log entry.
+ * Defined here as the single source of truth so the factory methods below
and the tests that
+ * assert on them reuse the same identifiers instead of duplicating the
string literals.
+ */
+ val LOG_WRITE_FAILURE_ERROR_CLASS = "STREAMING_CHECKPOINT_LOG_WRITE_FAILURE"
+ val COMMIT_LOG_WRITE_FAILURE = s"$LOG_WRITE_FAILURE_ERROR_CLASS.COMMIT_LOG"
+ val OFFSET_LOG_WRITE_FAILURE = s"$LOG_WRITE_FAILURE_ERROR_CLASS.OFFSET_LOG"
+
+ def commitLogWriteFailure(
+ batchId: Long,
+ checkpointLocation: String,
+ cause: Throwable): Throwable = {
+ new SparkException(
+ errorClass = COMMIT_LOG_WRITE_FAILURE,
+ messageParameters = Map(
+ "batchId" -> batchId.toString,
+ "checkpointLocation" -> checkpointLocation),
+ cause = cause
+ )
+ }
+
+ def offsetLogWriteFailure(
+ batchId: Long,
+ checkpointLocation: String,
+ cause: Throwable): Throwable = {
+ new SparkException(
+ errorClass = OFFSET_LOG_WRITE_FAILURE,
+ messageParameters = Map(
+ "batchId" -> batchId.toString,
+ "checkpointLocation" -> checkpointLocation),
+ cause = cause
+ )
+ }
+
def cannotLoadCheckpointFileManagerClass(path: String, className: String,
err: Throwable):
Throwable = {
new SparkException(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/AsyncProgressTrackingMicroBatchExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/AsyncProgressTrackingMicroBatchExecution.scala
index 156f2b8b169a..d779f48f5e6e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/AsyncProgressTrackingMicroBatchExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/AsyncProgressTrackingMicroBatchExecution.scala
@@ -20,11 +20,12 @@ package org.apache.spark.sql.execution.streaming.runtime
import java.util.concurrent._
import java.util.concurrent.atomic.AtomicLong
+import org.apache.spark.{SparkIllegalArgumentException, SparkRuntimeException,
SparkThrowable}
import org.apache.spark.internal.LogKeys.{BATCH_ID, PRETTY_ID_STRING}
import org.apache.spark.sql.catalyst.streaming.WriteToStream
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.execution.streaming.{AvailableNowTrigger,
OneTimeTrigger, ProcessingTimeTrigger}
+import org.apache.spark.sql.execution.streaming.{AvailableNowTrigger,
OneTimeTrigger, ProcessingTimeTrigger, RealTimeTrigger, StreamingErrors}
import org.apache.spark.sql.execution.streaming.checkpointing.{AsyncCommitLog,
AsyncOffsetSeqLog, CommitMetadata, OffsetSeqBase, OffsetSeqLog}
import
org.apache.spark.sql.execution.streaming.operators.stateful.StateStoreWriter
import org.apache.spark.sql.streaming.Trigger
@@ -43,8 +44,8 @@ class AsyncProgressTrackingMicroBatchExecution(
import AsyncProgressTrackingMicroBatchExecution._
- protected val asyncProgressTrackingCheckpointingIntervalMs: Long
- = getAsyncProgressTrackingCheckpointingIntervalMs(extraOptions)
+ protected val asyncProgressTrackingCheckpointingIntervalMs: Long =
+ getAsyncProgressTrackingCheckpointingIntervalMs(extraOptions, trigger)
// Offsets that are ready to be committed by the source.
// This is needed so that we can call source commit in the same thread as
micro-batch execution
@@ -61,8 +62,15 @@ class AsyncProgressTrackingMicroBatchExecution(
// writes to execute in order in a serialized fashion
protected val asyncWritesExecutorService
= ThreadUtils.newDaemonSingleThreadExecutorWithRejectedExecutionHandler(
- "async-log-write",
- 2, // one for offset commit and one for completion commit
+ AsyncProgressTrackingMicroBatchExecution.ASYNC_LOG_WRITE_THREAD_NAME,
+ if (trigger.isInstanceOf[RealTimeTrigger]) {
+ // Queue capacity 1; with the active task, at most 2 writes are in
flight. This bounds
+ // buffering so the completion commit of the last batch finishes before
the offset commit
+ // of the new batch starts.
+ 1
+ } else {
+ 2 // One for offset commit and one for completion commit.
+ },
new RejectedExecutionHandler() {
override def rejectedExecution(r: Runnable, executor:
ThreadPoolExecutor): Unit = {
try {
@@ -138,50 +146,77 @@ class AsyncProgressTrackingMicroBatchExecution(
// Hence, state repartitioning cannot happen.
}
- /**
- * Should not call super method as we need to do something completely
different
- * in this method for async progress tracking
- */
- override def markMicroBatchStart(execCtx: MicroBatchExecutionContext): Unit
= {
+ private def interruptStreamExecutionWithError(th: Throwable): Unit = {
+ logError(log"Interrupting stream execution due to error in async " +
+ log"progress tracking for query ${MDC(PRETTY_ID_STRING,
prettyIdString)}", th)
+ errorNotifier.markError(th)
+
+ // Immediately stop further processing.
+ sparkSession.sparkContext.cancelJobGroup(runId.toString)
+ // Cannot use stop() because of deadlock:
+ // stop() will not return until this async commit thread stops,
+ // but this thread cannot stop until stop() returns.
+ queryExecutionThread.interrupt()
+
+ // Codes after interrupt may never run, for critical cleanup, put them in
cleanup().
+ sparkSession.sparkContext.cancelJobGroup(runId.toString)
+ asyncWritesExecutorService.shutdownNow()
+ }
+
+ private def enqueueLastPersistedBatchForCommit(
+ batchId: Long, persistedToDurableStorage: Boolean): Unit = {
+ if (persistedToDurableStorage) {
+ // batch id cache not initialized
+ if (lastBatchPersistedToDurableStorage.get == -1) {
+ lastBatchPersistedToDurableStorage.set(
+ offsetLog.getPrevBatchFromStorage(batchId).getOrElse(-1))
+ }
+
+ if (batchId != 0 && lastBatchPersistedToDurableStorage.get != -1) {
+ // sanity check to make sure batch ids are monotonically increasing
+ assert(lastBatchPersistedToDurableStorage.get < batchId)
+ val prevBatchOff =
offsetLog.get(lastBatchPersistedToDurableStorage.get())
+ if (prevBatchOff.isDefined) {
+ // Offset is ready to be committed by the source. Add to queue
+ sourceCommitQueue.add(prevBatchOff.get)
+ } else {
+ throw new IllegalStateException(
+ s"Failed to commit processed data in the source because batch " +
+ s"${lastBatchPersistedToDurableStorage.get()} doesn't exist in
the offset log." +
+ s" This should not happen.")
+ }
+ }
+ lastBatchPersistedToDurableStorage.set(batchId)
+ }
+ }
+
+ private def asyncAddToOffsetLog(execCtx: MicroBatchExecutionContext): Unit =
{
// Because we are using a thread pool with only one thread, async writes
to the offset log
// are still written in a serial / in order fashion
- offsetLog
- .addAsync(execCtx.batchId,
- execCtx.endOffsets.toOffsets(sources, sourceIdMap,
execCtx.offsetSeqMetadata))
- .thenAccept((tuple: (Long, Boolean)) => {
- val (batchId: Long, persistedToDurableStorage: Boolean) = tuple
+ logInfo(log"Submitting async offset log write for batch ${MDC(BATCH_ID,
execCtx.batchId)}.")
+ val offsetSeq = execCtx.endOffsets.toOffsets(
+ sources, sourceIdMap, execCtx.offsetSeqMetadata)
+ offsetLog.addAsync(execCtx.batchId, offsetSeq)
+ .thenAccept { case (batchId, persistedToDurableStorage) =>
+ enqueueLastPersistedBatchForCommit(batchId, persistedToDurableStorage)
if (persistedToDurableStorage) {
- // batch id cache not initialized
- if (lastBatchPersistedToDurableStorage.get == -1) {
- lastBatchPersistedToDurableStorage.set(
- offsetLog.getPrevBatchFromStorage(batchId).getOrElse(-1L))
- }
-
- if (batchId != 0 && lastBatchPersistedToDurableStorage.get != -1) {
- // sanity check to make sure batch ids are monotonically increasing
- assert(lastBatchPersistedToDurableStorage.get < batchId)
- val prevBatchOff =
offsetLog.get(lastBatchPersistedToDurableStorage.get())
- if (prevBatchOff.isDefined) {
- // Offset is ready to be committed by the source. Add to queue
- sourceCommitQueue.add(prevBatchOff.get)
- } else {
- throw new IllegalStateException(
- s"Failed to commit processed data in the source because batch
" +
- s"${lastBatchPersistedToDurableStorage.get()} doesn't exist
in the offset log." +
- s" This should not happen.")
- }
- }
- lastBatchPersistedToDurableStorage.set(batchId)
+ logInfo(log"Committed async offset log to disk for batch
${MDC(BATCH_ID, batchId)}.")
+ } else {
+ logInfo(log"Committed async offset log to memory for batch
${MDC(BATCH_ID, batchId)}.")
}
- })
+ }
.exceptionally((th: Throwable) => {
logError(log"Encountered error while performing async offset write for
batch " +
log"${MDC(BATCH_ID, execCtx.batchId)}", th)
- errorNotifier.markError(th)
- return
+ handleAsyncLogWriteError(th, execCtx.batchId,
StreamingErrors.offsetLogWriteFailure)
+ null.asInstanceOf[Void]
})
+ }
- // check if there are offsets that are ready to be committed by the source
+ /**
+ * Check if there are offsets that are ready to be committed by the source.
+ */
+ private def commitAllSources(): Unit = {
var offset = sourceCommitQueue.poll()
while (offset != null) {
commitSources(offset)
@@ -189,40 +224,112 @@ class AsyncProgressTrackingMicroBatchExecution(
}
}
+ /**
+ * Should not call super method as we need to do something completely
different
+ * in this method for async progress tracking
+ */
+ override def markMicroBatchStart(execCtx: MicroBatchExecutionContext): Unit
= {
+ if (!trigger.isInstanceOf[RealTimeTrigger]) {
+ asyncAddToOffsetLog(execCtx)
+ }
+
+ commitAllSources()
+ }
+
override def markMicroBatchEnd(execCtx: MicroBatchExecutionContext): Unit = {
watermarkTracker.updateWatermark(execCtx.executionPlan.executedPlan)
+
+ if (trigger.isInstanceOf[RealTimeTrigger]) {
+ populateBatchOffsetsForRTM(execCtx)
+ execCtx.reportTimeTaken("walCommit") {
+ asyncAddToOffsetLog(execCtx)
+ }
+ }
+
+ // check if current batch there is a async write for the offset log is
issued for this batch
+ // if so, we should do the same for commit log. However, if this is the
first batch executed
+ // in this run we should always persist to the commit log. There can be
situations in which
+ // the offset log has more entries than the commit log and on restart we
need to make sure
+ // we write the missing entries to the commit log. For example if the
offset log is 0, 2, 5
+ // and the commit log is 0, 2. On restart we will re-process the data
from batch 3 -> 5.
+ // Batch 5 is already part of the offset log but we still need to write
the entry to
+ // the commit log
execCtx.reportTimeTaken("commitOffsets") {
- // check if current batch there is a async write for the offset log is
issued for this batch
- // if so, we should do the same for commit log. However, if this is the
first batch executed
- // in this run we should always persist to the commit log. There can be
situations in which
- // the offset log has more entries than the commit log and on restart we
need to make sure
- // we write the missing entries to the commit log. For example if the
offset log is 0, 2, 5
- // and the commit log is 0, 2. On restart we will re-process the data
from batch 3 -> 5.
- // Batch 5 is already part of the offset log but we still need to write
the entry to
- // the commit log
+ logInfo(log"Submitting async commit log write for batch ${MDC(BATCH_ID,
execCtx.batchId)}.")
+
if (offsetLog.getAsyncOffsetWrite(execCtx.batchId).nonEmpty
|| isFirstBatch) {
isFirstBatch = false
-
commitLog
.addAsync(execCtx.batchId,
CommitMetadata(watermarkTracker.currentWatermark))
+ .thenAccept((batchId: Long) => {
+ logInfo(log"Committed async commit log to disk for batch
${MDC(BATCH_ID, batchId)}.")
+ })
.exceptionally((th: Throwable) => {
logError(log"Got exception during async write to commit log for
batch " +
log"${MDC(BATCH_ID, execCtx.batchId)}", th)
- errorNotifier.markError(th)
- return
+ handleAsyncLogWriteError(th, execCtx.batchId,
StreamingErrors.commitLogWriteFailure)
+ null.asInstanceOf[Void]
})
} else {
if (!commitLog.addInMemory(
execCtx.batchId, CommitMetadata(watermarkTracker.currentWatermark)))
{
throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
}
+ logInfo(
+ log"Committed async commit log to memory for batch ${MDC(BATCH_ID,
execCtx.batchId)}.")
}
offsetLog.removeAsyncOffsetWrite(execCtx.batchId)
}
+ signalProcessAllAvailableIfRealTimeMode(execCtx)
committedOffsets ++= execCtx.endOffsets
}
+ /**
+ * Categorize a raw IO failure surfaced via an async log-write future, then
route it through
+ * the standard async error handling path. CompletableFuture wraps the
underlying cause in
+ * CompletionException, so unwrap before checking whether it has already
been categorized.
+ * `wrapAsLogWriteFailure` is one of the [[StreamingErrors]]
log-write-failure factories.
+ */
+ private def handleAsyncLogWriteError(
+ asyncWriteError: Throwable,
+ batchId: Long,
+ wrapAsLogWriteFailure: (Long, String, Throwable) => Throwable): Unit = {
+ val rootCause = asyncWriteError match {
+ case ce: CompletionException if ce.getCause != null => ce.getCause
+ case ee: ExecutionException if ee.getCause != null => ee.getCause
+ case other => other
+ }
+ val categorized = if (rootCause.isInstanceOf[SparkThrowable]) {
+ asyncWriteError
+ } else {
+ wrapAsLogWriteFailure(batchId, resolvedCheckpointRoot, rootCause)
+ }
+ if (trigger.isInstanceOf[RealTimeTrigger]) {
+ // For real time mode we need to interrupt the stream execution thread.
+ interruptStreamExecutionWithError(categorized)
+ } else {
+ errorNotifier.markError(categorized)
+ }
+ }
+
+ override protected def runActivatedStream(sparkSessionForStream:
SparkSession): Unit = {
+ try {
+ super.runActivatedStream(sparkSessionForStream)
+ } catch {
+ case e: Throwable
+ if StreamExecution.isInterruptionException(e,
sparkSession.sparkContext) &&
+ errorNotifier.getError().isDefined =>
+ throw new SparkRuntimeException(
+ errorClass = "STREAMING_ASYNC_OPERATION_FAILED",
+ messageParameters = Map("message" ->
errorNotifier.getError().get.getMessage),
+ cause = errorNotifier.getError().get
+ )
+ case e: Throwable =>
+ throw e
+ }
+ }
+
// need to look at the number of files on disk
override def purge(threshold: Long): Unit = {
while (offsetLog.writtenToDurableStorage.size() > minLogEntriesToMaintain)
{
@@ -237,11 +344,26 @@ class AsyncProgressTrackingMicroBatchExecution(
}
override def cleanup(): Unit = {
+ if (trigger.isInstanceOf[RealTimeTrigger] &&
errorNotifier.getError().isDefined) {
+ // No pending tasks in the queue should be executed if there was an
error.
+ asyncWritesExecutorService.shutdownNow()
+ // Waiting for at max 30s, which is aligned with ThreadUtils.shutdown.
+ asyncWritesExecutorService.awaitTermination(30, TimeUnit.SECONDS)
+ sparkSession.sparkContext.cancelJobGroup(runId.toString)
+ } else {
+ ThreadUtils.shutdown(asyncWritesExecutorService)
+ }
+ if (asyncWritesExecutorService.isShutdown) {
+ logInfo(log"Async progress tracking executor for query " +
+ log"${MDC(PRETTY_ID_STRING, prettyIdString)} has been shutdown")
+ } else {
+ logWarning(log"Async progress tracking executor for query " +
+ log"${MDC(PRETTY_ID_STRING, prettyIdString)} failed to shutdown
properly")
+ throw new TimeoutException(
+ "Failed to shutdown Async Progress Tracking executor in 30 seconds")
+ }
+ commitAllSources()
super.cleanup()
-
- ThreadUtils.shutdown(asyncWritesExecutorService)
- logInfo(log"Async progress tracking executor pool for query " +
- log"${MDC(PRETTY_ID_STRING, prettyIdString)} has been shutdown")
}
// used for testing
@@ -254,8 +376,7 @@ class AsyncProgressTrackingMicroBatchExecution(
private def validateAndGetTrigger(): TriggerExecutor = {
// validate that the pipeline is using a supported sink
if (!extraOptions
- .getOrElse(
- ASYNC_PROGRESS_TRACKING_OVERRIDE_SINK_SUPPORT_CHECK, "false")
+ .getOrElse(ASYNC_PROGRESS_TRACKING_OVERRIDE_SINK_SUPPORT_CHECK, "false")
.toBoolean) {
try {
plan.sink.name() match {
@@ -263,6 +384,8 @@ class AsyncProgressTrackingMicroBatchExecution(
case "console" =>
case "MemorySink" =>
case "KafkaTable" =>
+ case "ForeachSink" =>
+ case "ContinuousMemorySink" =>
case _ =>
throw new IllegalArgumentException(
s"Sink ${plan.sink.name()}" +
@@ -285,6 +408,13 @@ class AsyncProgressTrackingMicroBatchExecution(
trigger match {
case t: ProcessingTimeTrigger => ProcessingTimeExecutor(t, triggerClock)
+ case _: RealTimeTrigger =>
+ if (asyncProgressTrackingCheckpointingIntervalMs != 0) {
+ throw new SparkIllegalArgumentException(
+
"STREAMING_REAL_TIME_MODE.ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_NON_ZERO"
+ )
+ }
+ ProcessingTimeExecutor(ProcessingTimeTrigger(0), triggerClock)
case OneTimeTrigger =>
throw new IllegalArgumentException(
"Async progress tracking cannot be used with Once trigger")
@@ -313,16 +443,25 @@ object AsyncProgressTrackingMicroBatchExecution {
val ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_MS =
"asyncProgressTrackingCheckpointIntervalMs"
+ // Thread-name prefix of the single-threaded executor that performs async
offset/commit
+ // log writes. Tests match on this to distinguish async writes from
synchronous ones.
+ val ASYNC_LOG_WRITE_THREAD_NAME = "async-log-write"
+
// for testing purposes
val ASYNC_PROGRESS_TRACKING_OVERRIDE_SINK_SUPPORT_CHECK =
"_asyncProgressTrackingOverrideSinkSupportCheck"
private def getAsyncProgressTrackingCheckpointingIntervalMs(
- extraOptions: Map[String, String]): Long = {
+ extraOptions: Map[String, String],
+ trigger: Trigger): Long = {
extraOptions
.getOrElse(
ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_MS,
- "1000"
+ if (trigger.isInstanceOf[RealTimeTrigger]) {
+ "0"
+ } else {
+ "1000"
+ }
)
.toLong
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
index 7cfbb9669491..de84c9d15a6e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
@@ -1435,40 +1435,12 @@ class MicroBatchExecution(
updateStateStoreCkptId(execCtx, latestExecPlan)
}
- var needSignalProgressLock = false
// In real-time mode, we delay the offset logging until the end of the
batch.
// We first gather the offsets processed up to from all
RealTimeStreamScanExec,
// i.e. tasks that execute a source partition. We merge the offsets and
// write them to the offset log
if (trigger.isInstanceOf[RealTimeTrigger]) {
- val execs = StreamingQueryPlanTraverseHelper
- .collectFromUnfoldedPlan(lastExecution.executedPlan) {
- case e: RealTimeStreamScanExec => e
- }
-
- val endOffsetMap = MutableMap[SparkDataStream, OffsetV2]()
- execs.foreach { e =>
- val lowLatencyExec = e.asInstanceOf[RealTimeStreamScanExec]
- val accus: Seq[PartitionOffsetWithIndex] =
- lowLatencyExec.endOffsetsAccumulator.value.asScala.toSeq
- val sortedPartitionOffsets =
accus.sortBy(_.index).map(_.partitionOffset).toArray
- val source = e.stream
- val endOffset = source
- .asInstanceOf[SupportsRealTimeMode]
- .mergeOffsets(sortedPartitionOffsets)
- endOffsetMap += (source -> endOffset)
- }
-
- assert(endOffsetMap.size == execs.size, "Identical sources exist in the
physical nodes" +
- " which is not supported.")
-
- execCtx.endOffsets ++= endOffsetMap
- execCtx.recordEndOffsets(execCtx.endOffsets)
- execCtx.recordTriggerOffsets(
- from = execCtx.startOffsets,
- to = execCtx.endOffsets,
- latest = execCtx.latestOffsets
- )
+ populateBatchOffsetsForRTM(execCtx)
execCtx.reportTimeTaken("walCommit") {
if (!offsetLog.add(
execCtx.batchId,
@@ -1481,23 +1453,6 @@ class MicroBatchExecution(
log"Committed offsets for batch ${MDC(LogKeys.BATCH_ID,
execCtx.batchId)}. Metadata " +
log"${MDC(LogKeys.OFFSET_SEQUENCE_METADATA,
execCtx.offsetSeqMetadata)}"
)
- var shouldUpdate = true
- sources.foreach { s =>
- execCtx.startOffsets.get(s).foreach { prevOffsets =>
- if (!prevOffsets.equals(endOffsetMap(s))) {
- shouldUpdate = false
- }
- }
- }
- if (shouldUpdate) {
- // To trigger processAllAvailable() return.
- noNewData = true
-
- // We could signal ProcessAllAvailable to finish here, however
- // signaling after commit log will make it less likely that the caller
of
- // ProcessAllAvailable() sees offset log written but not commit log.
- needSignalProgressLock = true
- }
}
execCtx.reportTimeTaken("commitOffsets") {
@@ -1540,14 +1495,74 @@ class MicroBatchExecution(
throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
}
}
+ signalProcessAllAvailableIfRealTimeMode(execCtx)
committedOffsets ++= execCtx.endOffsets
+ }
+
+ /**
+ * When running in RealTimeTrigger mode, this method extracts the end
offsets after the current
+ * batch has finished execution and writes them to the execution context. In
real-time mode the
+ * end offsets are not known up front; they are gathered from all
[[RealTimeStreamScanExec]]
+ * nodes (i.e. tasks that execute a source partition), merged, and recorded
on the context so
+ * they can subsequently be written to the offset log.
+ */
+ protected def populateBatchOffsetsForRTM(execCtx:
MicroBatchExecutionContext): Unit = {
+ assert(trigger.isInstanceOf[RealTimeTrigger])
- // RealTime Mode deals with ProcessAllAvailable() differently. It sets
noNewData above
- // when a batch ends, so we need to signal here. Non-Real-Time mode sets
the same flag
- // in query planning phase.
- if (needSignalProgressLock) {
- withProgressLocked {
- awaitProgressLockCondition.signalAll()
+ val execs = StreamingQueryPlanTraverseHelper
+ .collectFromUnfoldedPlan(lastExecution.executedPlan) {
+ case e: RealTimeStreamScanExec => e
+ }
+
+ val endOffsetMap = MutableMap[SparkDataStream, OffsetV2]()
+ execs.foreach { e =>
+ val lowLatencyExec = e.asInstanceOf[RealTimeStreamScanExec]
+ val accus: Seq[PartitionOffsetWithIndex] =
+ lowLatencyExec.endOffsetsAccumulator.value.asScala.toSeq
+ val sortedPartitionOffsets =
accus.sortBy(_.index).map(_.partitionOffset).toArray
+ val source = e.stream
+ val endOffset = source
+ .asInstanceOf[SupportsRealTimeMode]
+ .mergeOffsets(sortedPartitionOffsets)
+ endOffsetMap += (source -> endOffset)
+ }
+
+ assert(endOffsetMap.size == execs.size, "Identical sources exist in the
physical nodes" +
+ " which is not supported.")
+
+ execCtx.endOffsets ++= endOffsetMap
+ execCtx.recordEndOffsets(execCtx.endOffsets)
+ execCtx.recordTriggerOffsets(
+ from = execCtx.startOffsets,
+ to = execCtx.endOffsets,
+ latest = execCtx.latestOffsets
+ )
+ }
+
+ /**
+ * RealTime Mode deals with ProcessAllAvailable() differently. Unlike
non-real-time mode, which
+ * sets [[noNewData]] in the query planning phase, real-time mode only knows
a batch was empty
+ * (i.e. the end offsets did not advance past the start offsets) after the
batch has finished.
+ * When that is the case, this signals the progress lock so a pending
processAllAvailable() call
+ * can return.
+ */
+ protected def signalProcessAllAvailableIfRealTimeMode(
+ execCtx: MicroBatchExecutionContext): Unit = {
+ if (trigger.isInstanceOf[RealTimeTrigger]) {
+ var emptyBatch = true
+ sources.foreach { s =>
+ execCtx.startOffsets.get(s).foreach { prevOffsets =>
+ if (!prevOffsets.equals(execCtx.endOffsets.get(s).get)) {
+ emptyBatch = false
+ }
+ }
+ }
+ if (emptyBatch) {
+ // To trigger processAllAvailable() return.
+ noNewData = true
+ withProgressLocked {
+ awaitProgressLockCondition.signalAll()
+ }
}
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/AsyncProgressTrackingRealTimeModeSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/AsyncProgressTrackingRealTimeModeSuite.scala
new file mode 100644
index 000000000000..453072245fb3
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/AsyncProgressTrackingRealTimeModeSuite.scala
@@ -0,0 +1,381 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.File
+
+import org.scalatest.BeforeAndAfter
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.time.{Seconds, Span}
+
+import org.apache.spark.{SparkException, SparkRuntimeException}
+import
org.apache.spark.sql.execution.streaming.runtime.{AsyncProgressTrackingMicroBatchExecution,
StreamExecution}
+import
org.apache.spark.sql.execution.streaming.runtime.AsyncProgressTrackingMicroBatchExecution.{ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_MS,
ASYNC_PROGRESS_TRACKING_ENABLED}
+import org.apache.spark.sql.execution.streaming.runtime.MicroBatchExecution
+import org.apache.spark.sql.execution.streaming.sources.ContinuousMemorySink
+import org.apache.spark.sql.execution.streaming.sources.LowLatencyMemoryStream
+import org.apache.spark.sql.execution.streaming.sources.MemorySink
+import
org.apache.spark.sql.execution.streaming.state.{FailureInjectionCheckpointFileManager,
FailureInjectionFileSystem}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.streaming.{OutputMode,
StreamRealTimeModeManualClockSuiteBase}
+import org.apache.spark.util.Utils
+
+
+private class UnsupportedSink extends MemorySink {
+ override def name(): String = "UnsupportedSink"
+}
+
+class AsyncProgressTrackingRealTimeModeSuite
+ extends StreamRealTimeModeManualClockSuiteBase with BeforeAndAfter with
Matchers {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ def waitPendingOffsetWrites(streamExecution: StreamExecution): Unit = {
+
assert(streamExecution.isInstanceOf[AsyncProgressTrackingMicroBatchExecution])
+ eventually(timeout(Span(60, Seconds))) {
+ streamExecution
+ .asInstanceOf[AsyncProgressTrackingMicroBatchExecution]
+ .areWritesPendingOrInProgress() should be(false)
+ }
+ }
+
+ def getListOfFiles(dir: String): List[File] = {
+ val d = new File(dir)
+ if (d.exists && d.isDirectory) {
+ d.listFiles.filter(_.isFile).toList
+ } else {
+ List[File]()
+ }
+ }
+
+ def getBatchIdsSortedFromLog(logPath: String): List[Int] = {
+ getListOfFiles(logPath)
+ .filter(file => !file.isHidden)
+ .map(file => file.getName.toInt)
+ .sorted
+ }
+
+ test("async progress tracking happy path") {
+ val checkpointLocation = Utils.createTempDir(namePrefix =
"streaming.metadata").getCanonicalPath
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+
+ testStream(
+ df,
+ outputMode = OutputMode.Update(),
+ sink = new ContinuousMemorySink(),
+ extraOptions = Map(
+ ASYNC_PROGRESS_TRACKING_ENABLED -> "true",
+ ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_MS -> "0"
+ ))(
+ StartStream(checkpointLocation = checkpointLocation),
+ AddData(inputData, 0 until 10: _*),
+ CheckAnswerWithTimeout(60000, 0 until 10: _*),
+ advanceRealTimeClock,
+ AddData(inputData, 10 until 20: _*),
+ CheckAnswerWithTimeout(60000, 0 until 20: _*),
+ advanceRealTimeClock,
+ AddData(inputData, 20 until 30: _*),
+ CheckAnswerWithTimeout(60000, 0 until 30: _*),
+ advanceRealTimeClock,
+ WaitUntilBatchProcessed(2),
+ Execute { query =>
+ waitPendingOffsetWrites(query)
+ getBatchIdsSortedFromLog(checkpointLocation + "/offsets") should
+ equal(Array(0, 1, 2))
+ getBatchIdsSortedFromLog(checkpointLocation + "/commits") should
+ equal(Array(0, 1, 2))
+ }
+ )
+ }
+
+ test("async progress tracking enabled by default in RTM for stateless
queries") {
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+ testStream(df, outputMode = OutputMode.Update)(
+ AssertOnQuery(_.getClass ==
classOf[AsyncProgressTrackingMicroBatchExecution]))
+ }
+
+ test("user setting for APT takes precedence over default behavior") {
+ // We respect the user setting even if the query is APT-compatible
+ {
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+ testStream(
+ df,
+ outputMode = OutputMode.Update,
+ extraOptions = Map(ASYNC_PROGRESS_TRACKING_ENABLED -> "false"))(
+ AssertOnQuery(_.getClass == classOf[MicroBatchExecution]))
+ }
+
+ {
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+ testStream(
+ df,
+ outputMode = OutputMode.Update,
+ extraOptions = Map(ASYNC_PROGRESS_TRACKING_ENABLED -> "true"))(
+ AssertOnQuery(_.getClass ==
classOf[AsyncProgressTrackingMicroBatchExecution]))
+ }
+
+ // Even if it would lead to an error
+ {
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+ val e = intercept[IllegalArgumentException] {
+ testStream(
+ df,
+ outputMode = OutputMode.Update,
+ sink = new UnsupportedSink(),
+ extraOptions = Map(ASYNC_PROGRESS_TRACKING_ENABLED -> "true"))()
+ }
+ assert(e.getMessage.contains("does not support async progress tracking"))
+ }
+ }
+
+ test("we do not enable APT in RTM by default when the logical plan is
stateful") {
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF().groupBy("value").count()
+ testStream(df, outputMode = OutputMode.Update)(
+ AssertOnQuery(_.getClass !=
classOf[AsyncProgressTrackingMicroBatchExecution]))
+ }
+
+ test("async progress tracking recovery") {
+ val checkpointLocation = Utils.createTempDir(namePrefix =
"streaming.metadata").getCanonicalPath
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+ val sink = new ContinuousMemorySink()
+
+ testStream(df, outputMode = OutputMode.Update(), sink = sink, extraOptions
= Map(
+ ASYNC_PROGRESS_TRACKING_ENABLED -> "true",
+ ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_MS -> "0"
+ ))(
+ StartStream(checkpointLocation = checkpointLocation),
+ AddData(inputData, 0 until 10: _*),
+ CheckAnswerWithTimeout(60000, 0 until 10: _*),
+ advanceRealTimeClock,
+ AddData(inputData, 10 until 20: _*),
+ CheckAnswerWithTimeout(60000, 0 until 20: _*),
+ advanceRealTimeClock,
+ WaitUntilBatchProcessed(1),
+ StopStream, // Should automatically wait for pending writes.
+ Execute { _ =>
+ sink.clear()
+ getBatchIdsSortedFromLog(checkpointLocation + "/offsets") should
+ equal(Array(0, 1))
+ getBatchIdsSortedFromLog(checkpointLocation + "/commits") should
+ equal(Array(0, 1))
+ },
+ // Restart the query.
+ StartStream(checkpointLocation = checkpointLocation),
+ AddData(inputData, 100),
+ CheckAnswerWithTimeout(60000, 100) // Should not reprocess old data.
+ )
+ }
+
+ test("restart a sync-checkpoint RTM query with async progress tracking on by
default") {
+ // Enabling APT by default for stateless RTM means an existing query that
previously ran on
+ // the synchronous MicroBatchExecution will silently switch to
+ // AsyncProgressTrackingMicroBatchExecution on its next restart. This
verifies the async engine
+ // can resume from a checkpoint written by the sync engine, and the
reverse direction, without
+ // reprocessing or skipping data. A fresh sink is used per run so each
assertion sees only that
+ // run's output.
+ val checkpointLocation = Utils.createTempDir(namePrefix =
"streaming.metadata").getCanonicalPath
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+
+ // Run 1: force the synchronous engine by disabling APT explicitly.
+ testStream(df, outputMode = OutputMode.Update(), sink = new
ContinuousMemorySink(),
+ extraOptions = Map(ASYNC_PROGRESS_TRACKING_ENABLED -> "false"))(
+ StartStream(checkpointLocation = checkpointLocation),
+ AssertOnQuery(_.getClass == classOf[MicroBatchExecution]),
+ AddData(inputData, 0 until 10: _*),
+ CheckAnswerWithTimeout(60000, 0 until 10: _*),
+ advanceRealTimeClock,
+ WaitUntilBatchProcessed(0),
+ StopStream
+ )
+
+ // Run 2: same checkpoint, default settings -> APT is now enabled. Must
resume, not replay.
+ testStream(df, outputMode = OutputMode.Update(), sink = new
ContinuousMemorySink())(
+ StartStream(checkpointLocation = checkpointLocation),
+ AssertOnQuery(_.getClass ==
classOf[AsyncProgressTrackingMicroBatchExecution]),
+ AddData(inputData, 10 until 20: _*),
+ CheckAnswerWithTimeout(60000, 10 until 20: _*), // Only new data; 0..9
must not reprocess.
+ advanceRealTimeClock,
+ WaitUntilBatchProcessed(1),
+ StopStream
+ )
+
+ // Run 3: restart again with APT explicitly disabled to confirm the
reverse direction
+ // (APT-written checkpoint resumed by the sync engine) also resumes
cleanly.
+ testStream(df, outputMode = OutputMode.Update(), sink = new
ContinuousMemorySink(),
+ extraOptions = Map(ASYNC_PROGRESS_TRACKING_ENABLED -> "false"))(
+ StartStream(checkpointLocation = checkpointLocation),
+ AssertOnQuery(_.getClass == classOf[MicroBatchExecution]),
+ AddData(inputData, 20 until 30: _*),
+ CheckAnswerWithTimeout(60000, 20 until 30: _*), // Only new data;
nothing reprocessed.
+ advanceRealTimeClock,
+ StopStream
+ )
+ }
+
+ test("current batch wait for the previous batch to commit") {
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key ->
+ classOf[FailureInjectionCheckpointFileManager].getName) {
+ withTempDir { checkpointDir =>
+ val injectionState =
FailureInjectionFileSystem.registerTempPath(checkpointDir.getPath)
+ try {
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+
+ // This regex will delay the createAtomic() of commit file for batch
1.
+ injectionState.delayCreateAtomicRegex = Seq(".*/commits/1")
+
+ testStream(df, outputMode = OutputMode.Update(), sink = new
ContinuousMemorySink(),
+ extraOptions = Map(
+ ASYNC_PROGRESS_TRACKING_ENABLED -> "true",
+ ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_MS -> "0"
+ ))(
+ StartStream(checkpointLocation = checkpointDir.getAbsolutePath),
+
+ // Batch 0 - should complete normally.
+ AddData(inputData, 0 until 10: _*),
+ CheckAnswerWithTimeout(60000, 0 until 10: _*),
+ advanceRealTimeClock,
+ WaitUntilBatchProcessed(0),
+
+ // Batch 1 - the commit will be delayed!
+ AddData(inputData, 10 until 20: _*),
+ CheckAnswerWithTimeout(60000, 0 until 20: _*),
+ advanceRealTimeClock,
+ WaitUntilBatchProcessed(1),
+
+ // Batch 2 - should wait for batch 1's commit to complete.
+ AddData(inputData, 20 until 30: _*),
+ CheckAnswerWithTimeout(60000, 0 until 30: _*),
+ advanceRealTimeClock,
+ // At this point batch 2 should be blocked waiting for batch 1's
commit.
+
+ // Verify that batch 2 is waiting.
+ Execute { query =>
+ Thread.sleep(2000) // wait a bit to ensure batch 2 is blocked
+ query.lastProgress.batchId should be(1) // still at batch 1
+ },
+
+ // Signal the delayed streams to allow batch 1's commit to finish,
and unblock batch 2.
+ Execute { q =>
+ FailureInjectionFileSystem.createAtomicDelaySemaphore.release()
+ // Clear the regex so subsequent batches aren't delayed.
+ injectionState.delayCreateAtomicRegex = Seq.empty
+ },
+ WaitUntilBatchProcessed(2),
+ Execute { query =>
+ waitPendingOffsetWrites(query)
+ getBatchIdsSortedFromLog(checkpointDir.getAbsolutePath +
"/offsets") should
+ equal(Array(0, 1, 2))
+ getBatchIdsSortedFromLog(checkpointDir.getAbsolutePath +
"/commits") should
+ equal(Array(0, 1, 2))
+ },
+ StopStream
+ )
+ } finally {
+
FailureInjectionFileSystem.removePathFromTempToInjectionState(checkpointDir.getPath)
+ }
+ }
+ }
+ }
+
+ // Inject a close() failure on the batch-1 write of the given log ("commits"
or "offsets") and
+ // assert the stream fails fast with the matching categorized
log-write-failure error, preserving
+ // the original IOException as the cause. Covers both the commit-log and
offset-log async write
+ // failure paths in RTM.
+ private def testAsyncLogWriteFailure(logType: String, expectedErrorClass:
String): Unit = {
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key ->
+ classOf[FailureInjectionCheckpointFileManager].getName) {
+ withTempDir { checkpointDir =>
+ val injectionState =
FailureInjectionFileSystem.registerTempPath(checkpointDir.getPath)
+ try {
+ val inputData = LowLatencyMemoryStream[Int]
+ val df = inputData.toDF()
+
+ // This regex will make the close() of the batch-1 log file for
`logType` throw error.
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/$logType/1")
+
+ testStream(df, outputMode = OutputMode.Update(), sink = new
ContinuousMemorySink(),
+ extraOptions = Map(
+ ASYNC_PROGRESS_TRACKING_ENABLED -> "true",
+ ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_MS -> "0"
+ ))(
+ StartStream(checkpointLocation = checkpointDir.getAbsolutePath),
+
+ // Batch 0 - should complete normally.
+ AddData(inputData, 0 until 10: _*),
+ CheckAnswerWithTimeout(60000, 0 until 10: _*),
+ advanceRealTimeClock,
+ WaitUntilBatchProcessed(0),
+
+ // Batch 1 - The close() will throw IOException, and the stream
should fail fast.
+ AddData(inputData, 10 until 20: _*),
+ CheckAnswerWithTimeout(60000, 0 until 20: _*),
+ advanceRealTimeClock,
+ ExpectFailure[Exception](
+ assertFailure = e => {
+ // The async log write failure surfaces in one of two valid
ways, depending on a
+ // benign race in how the execution thread observes the error:
either wrapped in
+ // STREAMING_ASYNC_OPERATION_FAILED (when the thread is
interrupted mid-batch) or
+ // thrown directly (when the post-batch error check picks it
up first). Unwrap the
+ // async wrapper if present, then assert on the categorized
log write failure.
+ val categorized = e match {
+ case sre: SparkRuntimeException
+ if sre.getCondition ==
"STREAMING_ASYNC_OPERATION_FAILED" => sre.getCause
+ case other => other
+ }
+ val sparkEx = categorized.asInstanceOf[SparkException]
+ checkError(
+ sparkEx,
+ expectedErrorClass,
+ parameters = Map("batchId" -> "1", "checkpointLocation" ->
".*"),
+ matchPVals = true)
+ // The original IOException must be preserved as the root
cause.
+ assert(sparkEx.getCause.isInstanceOf[java.io.IOException])
+ assert(sparkEx.getCause.getMessage === "Fake File Stream Close
Failure")
+ },
+ typeIsSuperClass = true)
+ )
+ } finally {
+
FailureInjectionFileSystem.removePathFromTempToInjectionState(checkpointDir.getPath)
+ }
+ }
+ }
+ }
+
+ test("throw exceptions immediately on commit log write failure") {
+ testAsyncLogWriteFailure("commits",
StreamingErrors.COMMIT_LOG_WRITE_FAILURE)
+ }
+
+ test("throw exceptions immediately on offset log write failure") {
+ testAsyncLogWriteFailure("offsets",
StreamingErrors.OFFSET_LOG_WRITE_FAILURE)
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala
index 898e324a954e..43e90a295c36 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala
@@ -70,8 +70,15 @@ class FailureInjectionCheckpointFileManager(path: Path,
hadoopConf: Configuratio
// Injection state for the path
private val injectionState =
FailureInjectionFileSystem.getInjectionState(path.toString)
- override def createAtomic(path: Path,
- overwriteIfPossible: Boolean):
CancellableFSDataOutputStream = {
+ override def createAtomic(
+ path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ injectionState.delayCreateAtomicRegex.foreach { pattern =>
+ if (path.toString.matches(pattern)) {
+ // wait on the semaphore before proceeding
+ FailureInjectionFileSystem.createAtomicDelaySemaphore.acquire()
+ }
+ }
injectionState.failureCreateAtomicRegex.foreach { pattern =>
if (path.toString.matches(pattern)) {
throw new IOException("Fake File System Create Atomic Failure")
@@ -128,6 +135,10 @@ class FailureInjectionState {
// File names matching this regex will cause the createAtomic() to fail
@volatile
var failureCreateAtomicRegex: Seq[String] = Seq.empty
+ // File names matching this regex will cause the createAtomic() to delay on
+ // createAtomicDelaySemaphore
+ @volatile
+ var delayCreateAtomicRegex: Seq[String] = Seq.empty
// If true, Exists() call will fail
@volatile
var shouldFailExist: Boolean = false
@@ -150,6 +161,9 @@ object FailureInjectionFileSystem {
// A map from a temp path to its failure injection state.
var tempPathToInjectionState: Map[String, FailureInjectionState] = Map.empty
+ // A semaphore to delay the return of createAtomic
+ val createAtomicDelaySemaphore = new java.util.concurrent.Semaphore(0)
+
/**
* Create a new FailureInjectionState for a temp path and add it to the map.
* @param path the temp path
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala
index cea70676bb7b..0f584d818c06 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeSuite.scala
@@ -148,22 +148,6 @@ class StreamRealTimeModeSuite extends
StreamRealTimeModeSuiteBase {
)
}
- test("environment check for real-time mode throws when the valid
configurations aren't set") {
- val inputData = LowLatencyMemoryStream.singlePartition[Int]
- val mapped = inputData.toDS().map(_ + 1)
-
- checkError(
- intercept[SparkIllegalArgumentException] {
- testStream(mapped, OutputMode.Update, Map(
- "asyncProgressTrackingEnabled" -> "true"
- ), new ContinuousMemorySink())(
- StartStream()
- )
- },
- "STREAMING_REAL_TIME_MODE.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED"
- )
- }
-
test("error when unsupported source is used") {
val inputData = MemoryStream[Int]
val mapped = inputData.toDS().map(_ + 1)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala
index a6067aaf189e..61864f57c8ad 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala
@@ -295,10 +295,12 @@ trait StreamTest extends SharedSparkSession with
TimeLimits {
*/
case class ExpectFailure[T <: Throwable : ClassTag](
assertFailure: Throwable => Unit = _ => {},
- isFatalError: Boolean = false) extends StreamAction {
+ isFatalError: Boolean = false,
+ typeIsSuperClass: Boolean = false) extends StreamAction {
val causeClass: Class[T] =
implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]]
override def toString(): String =
- s"ExpectFailure[${causeClass.getName}, isFatalError: $isFatalError]"
+ s"ExpectFailure[${causeClass.getName}, isFatalError: $isFatalError, " +
+ s"typeIsSuperClass: $typeIsSuperClass]"
}
/**
@@ -712,14 +714,25 @@ trait StreamTest extends SharedSparkSession with
TimeLimits {
s"incorrect exception returned by query.exception()")
val exception = currentStream.exception.get
- verify(exception.cause.getClass === ef.causeClass,
- "incorrect cause in exception returned by query.exception()\n" +
- s"\tExpected: ${ef.causeClass}\n\tReturned:
${exception.cause.getClass}")
+ if (ef.typeIsSuperClass) {
+ verify(ef.causeClass.isInstance(exception.cause),
+ "incorrect cause in exception returned by query.exception()\n"
+
+ s"\tExpected: ${ef.causeClass}\n\tReturned:
${exception.cause.getClass}")
+ } else {
+ verify(exception.cause.getClass === ef.causeClass,
+ "incorrect cause in exception returned by query.exception()\n"
+
+ s"\tExpected: ${ef.causeClass}\n\tReturned:
${exception.cause.getClass}")
+ }
if (ef.isFatalError) {
// This is a fatal error, `streamThreadDeathCause` should be set
to this error in
// UncaughtExceptionHandler.
- verify(streamThreadDeathCause != null &&
- streamThreadDeathCause.getClass === ef.causeClass,
+ val fatalErrorMatches = streamThreadDeathCause != null && (
+ if (ef.typeIsSuperClass) {
+ ef.causeClass.isInstance(streamThreadDeathCause)
+ } else {
+ streamThreadDeathCause.getClass === ef.causeClass
+ })
+ verify(fatalErrorMatches,
"UncaughtExceptionHandler didn't receive the correct error\n" +
s"\tExpected: ${ef.causeClass}\n\tReturned:
$streamThreadDeathCause")
streamThreadDeathCause = null
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
