This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 86d8e8128d0e [SPARK-59844][SS][RTM] Preserve root causes for RTM async
operation failures
86d8e8128d0e is described below
commit 86d8e8128d0eab5330c5f9194b5f4d7a0180f028
Author: Boyang Jerry Peng <[email protected]>
AuthorDate: Thu Oct 1 12:48:57 2026 +0900
[SPARK-59844][SS][RTM] Preserve root causes for RTM async operation failures
### What changes were proposed in this pull request?
This PR makes async progress tracking failures in Real-Time Mode report
the error that actually interrupted the query.
It:
- Tracks the first interruption-triggering async error separately from
the first error retained by `ErrorNotifier`.
- Recursively unwraps `CompletionException` and `ExecutionException`
layers while preserving an existing `SparkThrowable`.
- Produces the same `STREAMING_ASYNC_OPERATION_FAILED` cause chain
regardless of whether the query thread observes the interruption or the
notifier error first.
- Retains an independent earlier async error as suppressed context.
- Prevents `ErrorNotifier` from creating cyclic or overlapping
cause/suppressed graphs, including for non-RTM users.
- Preserves fatal-error behavior by avoiding wrapping fatal errors
observed during the notifier race.
The resulting primary exception chain is:
```text
STREAMING_ASYNC_OPERATION_FAILED
└── STREAMING_CHECKPOINT_LOG_WRITE_FAILURE.{OFFSET_LOG|COMMIT_LOG}
└── underlying storage exception
```
### Why are the changes needed?
Async progress tracking writes checkpoint offset and commit logs on a
background thread. In Real-Time Mode, a failure in that thread interrupts the
query execution thread.
ErrorNotifier retains the first asynchronous error it observes. If it had
already retained an earlier, non-interrupting error, the query could report
that earlier error instead of the
checkpoint-log failure that caused the interruption. A race with the
foreground error check could also make the reported exception shape depend on
timing.
Additionally, asynchronous futures can introduce nested
CompletionException or ExecutionException wrappers that obscure an
already-categorized Spark error. Attaching related exceptions
as suppressed errors could create cyclic or overlapping throwable graphs,
producing [CIRCULAR REFERENCE] in rendered stack traces.
This change makes the primary cause deterministic and exposes the
actionable checkpoint-storage exception while retaining independent diagnostic
context.
### Does this PR introduce any user-facing change?
No
### How was this patch tested?
Added and ran focused tests covering:
- First-error retention and conditional error publication in
ErrorNotifier.
- Concurrent suppression of independent errors.
- Concurrent errors with overlapping throwable graphs.
- RTM offset-log and commit-log write failures.
- An earlier retained error followed by an interruption-triggering error.
- Both interruption and notifier-race outcomes.
- Preservation of fatal errors.
- Nested CompletionException and ExecutionException wrappers.
- Categorized failures that are themselves ExecutionException instances.
- Throwable graphs rendered without [CIRCULAR REFERENCE].
The following suites passed:
- ErrorNotifierSuite: 8 tests
- AsyncProgressTrackingRealTimeModeSuite: 17 tests
build/sbt \
'core/testOnly org.apache.spark.util.ErrorNotifierSuite' \
'sql/testOnly
org.apache.spark.sql.execution.streaming.AsyncProgressTrackingRealTimeModeSuite'
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with OpenAI Codex (GPT-5)
Closes #59116 from jerrypeng/jerrypeng/streaming-async-root-cause.
Authored-by: Boyang Jerry Peng <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
(cherry picked from commit b1f606c2c0b1e10e4a6f493cd68cc80b87a7379d)
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../org/apache/spark/util/ErrorNotifier.scala | 58 ++++++-
.../org/apache/spark/util/ErrorNotifierSuite.scala | 48 ++++++
.../AsyncProgressTrackingMicroBatchExecution.scala | 97 ++++++++---
.../AsyncProgressTrackingRealTimeModeSuite.scala | 178 +++++++++++++++++++--
4 files changed, 336 insertions(+), 45 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/util/ErrorNotifier.scala
b/core/src/main/scala/org/apache/spark/util/ErrorNotifier.scala
index 18ed7688f40a..a40e0c7bd524 100644
--- a/core/src/main/scala/org/apache/spark/util/ErrorNotifier.scala
+++ b/core/src/main/scala/org/apache/spark/util/ErrorNotifier.scala
@@ -17,6 +17,7 @@
package org.apache.spark.util
+import java.util.{ArrayDeque, Collections, IdentityHashMap}
import java.util.concurrent.atomic.AtomicReference
import org.apache.spark.internal.Logging
@@ -36,20 +37,36 @@ import org.apache.spark.internal.Logging
private[spark] class ErrorNotifier extends Logging {
private val error = new AtomicReference[Throwable]
+ private val suppressionLock = new Object
/**
- * Record a fatal error. Only the first error is retained - subsequent calls
- * are no-ops so cascading failures cannot mask the original cause.
+ * Record a fatal error. Only the first error is retained. Subsequent errors
with disjoint
+ * throwable graphs are attached to it as suppressed exceptions so cascading
failures cannot
+ * mask the original cause or create cyclic exception chains.
*/
def markError(th: Throwable): Unit = {
+ if (!tryMarkError(th)) {
+ suppressionLock.synchronized {
+ val existing = error.get()
+ if (existing != null && existing != th &&
+ ErrorNotifier.haveDisjointThrowableGraphs(existing, th)) {
+ existing.addSuppressed(th)
+ }
+ }
+ }
+ }
+
+ /**
+ * Record an error only if none has been retained.
+ *
+ * @return true if this error was retained
+ */
+ def tryMarkError(th: Throwable): Boolean = {
if (error.compareAndSet(null, th)) {
logError(log"A fatal error has occurred.", th)
+ true
} else {
- // Attach subsequent errors as suppressed so they're not silently lost.
- val existing = error.get()
- if (existing != null && existing != th) {
- existing.addSuppressed(th)
- }
+ false
}
}
@@ -63,3 +80,30 @@ private[spark] class ErrorNotifier extends Logging {
getError().foreach({th => throw th})
}
}
+
+private[spark] object ErrorNotifier {
+
+ /**
+ * @return true if the cause and suppressed graphs share no Throwable object
by identity
+ */
+ def haveDisjointThrowableGraphs(first: Throwable, second: Throwable):
Boolean = {
+ def collect(root: Throwable): java.util.Set[Throwable] = {
+ val graph = Collections.newSetFromMap(
+ new IdentityHashMap[Throwable, java.lang.Boolean]())
+ val pending = new ArrayDeque[Throwable]()
+ pending.push(root)
+ while (!pending.isEmpty) {
+ val current = pending.pop()
+ if (graph.add(current)) {
+ if (current.getCause != null) {
+ pending.push(current.getCause)
+ }
+ current.getSuppressed.foreach(pending.push)
+ }
+ }
+ graph
+ }
+
+ Collections.disjoint(collect(first), collect(second))
+ }
+}
diff --git a/core/src/test/scala/org/apache/spark/util/ErrorNotifierSuite.scala
b/core/src/test/scala/org/apache/spark/util/ErrorNotifierSuite.scala
index 313857eb1de1..bfd4d680ab2d 100644
--- a/core/src/test/scala/org/apache/spark/util/ErrorNotifierSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/ErrorNotifierSuite.scala
@@ -63,6 +63,29 @@ class ErrorNotifierSuite extends SparkFunSuite {
assert(err.getSuppressed.isEmpty)
}
+ test("markError does not suppress an error whose graph contains the retained
error") {
+ val notifier = new ErrorNotifier
+ val first = new RuntimeException("first")
+ val second = new RuntimeException("second", first)
+
+ notifier.markError(first)
+ notifier.markError(second)
+
+ assert(notifier.getError().contains(first))
+ assert(first.getSuppressed.isEmpty)
+ }
+
+ test("tryMarkError does not suppress an error when another error was
retained first") {
+ val notifier = new ErrorNotifier
+ val first = new RuntimeException("first")
+ val second = new RuntimeException("second")
+
+ assert(notifier.tryMarkError(first))
+ assert(!notifier.tryMarkError(second))
+ assert(notifier.getError().contains(first))
+ assert(first.getSuppressed.isEmpty)
+ }
+
test("concurrent markError retains exactly one error and suppresses the
rest") {
val notifier = new ErrorNotifier
val numThreads = 8
@@ -84,4 +107,29 @@ class ErrorNotifierSuite extends SparkFunSuite {
// Every losing error must be attached as suppressed on the winner, so
none is lost.
assert(winner.get.getSuppressed.toSet == errors.filterNot(_ eq
winner.get).toSet)
}
+
+ test("concurrent markError suppresses only one error when losing graphs
overlap") {
+ val notifier = new ErrorNotifier
+ val first = new RuntimeException("first")
+ val sharedCause = new RuntimeException("shared")
+ val numThreads = 8
+ val errors = (0 until numThreads).map { i =>
+ new RuntimeException(s"err-$i", sharedCause)
+ }
+ val barrier = new CyclicBarrier(numThreads)
+ notifier.markError(first)
+
+ val threads = errors.map { error =>
+ new Thread(() => {
+ barrier.await()
+ notifier.markError(error)
+ })
+ }
+ threads.foreach(_.start())
+ threads.foreach(_.join())
+
+ assert(first.getSuppressed.length === 1)
+ assert(errors.contains(first.getSuppressed.head))
+ assert(!SparkErrorUtils.stackTraceToString(first).contains("CIRCULAR
REFERENCE"))
+ }
}
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 11bda6221582..ddb5d702fd9f 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
@@ -17,8 +17,12 @@
package org.apache.spark.sql.execution.streaming.runtime
+import java.util.{Collections, IdentityHashMap}
import java.util.concurrent._
-import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
+
+import scala.annotation.tailrec
+import scala.util.control.NonFatal
import org.apache.spark.{SparkIllegalArgumentException, SparkRuntimeException,
SparkThrowable}
import org.apache.spark.internal.LogKeys.{BATCH_ID, PRETTY_ID_STRING}
@@ -29,7 +33,7 @@ import
org.apache.spark.sql.execution.streaming.{AvailableNowTrigger, OneTimeTri
import org.apache.spark.sql.execution.streaming.checkpointing.{AsyncCommitLog,
AsyncOffsetSeqLog, CommitMetadataBase, OffsetSeqBase, OffsetSeqLog}
import
org.apache.spark.sql.execution.streaming.operators.stateful.StateStoreWriter
import org.apache.spark.sql.streaming.Trigger
-import org.apache.spark.util.{Clock, ThreadUtils}
+import org.apache.spark.util.{Clock, ErrorNotifier, ThreadUtils}
/**
* Class to execute micro-batches when async progress tracking is enabled
@@ -55,6 +59,10 @@ class AsyncProgressTrackingMicroBatchExecution(
// to cache the batch id of the last batch written to storage
private val lastBatchPersistedToDurableStorage = new AtomicLong(-1)
+ // Error that caused RTM to interrupt the query execution thread.
ErrorNotifier retains the
+ // first async error, which may instead be an earlier failure from a
non-interrupting operation.
+ private val interruptingAsyncError = new AtomicReference[Throwable]()
+
// used to check during the first batch if the pipeline is stateful
private var isFirstBatch: Boolean = true
@@ -149,7 +157,13 @@ class AsyncProgressTrackingMicroBatchExecution(
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)
+ if (interruptingAsyncError.compareAndSet(null, th)) {
+ // Do not attach the primary error to an earlier retained error. The
final wrapper points to
+ // both, and sharing the primary across those branches breaks
Throwable's rendered chain.
+ errorNotifier.tryMarkError(th)
+ } else {
+ errorNotifier.markError(th)
+ }
// Immediately stop further processing.
sparkSession.sparkContext.cancelJobGroup(runId.toString)
@@ -301,22 +315,35 @@ class AsyncProgressTrackingMicroBatchExecution(
}
/**
- * 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.
+ * Normalize and categorize a failure from an async log-write future, then
route it through the
+ * standard async error handling path.
+ *
+ * @param wrapAsLogWriteFailure factory for an uncategorized log-write
failure
*/
- private def handleAsyncLogWriteError(
+ private[streaming] 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 seenWrappers = Collections.newSetFromMap(
+ new IdentityHashMap[Throwable, java.lang.Boolean]())
+ @tailrec
+ def unwrapFutureFailure(error: Throwable): Throwable = {
+ if (!seenWrappers.add(error)) {
+ error
+ } else {
+ error match {
+ case _: SparkThrowable => error
+ case wrapper: CompletionException if wrapper.getCause != null =>
+ unwrapFutureFailure(wrapper.getCause)
+ case wrapper: ExecutionException if wrapper.getCause != null =>
+ unwrapFutureFailure(wrapper.getCause)
+ case other => other
+ }
+ }
}
+ val rootCause = unwrapFutureFailure(asyncWriteError)
val categorized = if (rootCause.isInstanceOf[SparkThrowable]) {
- asyncWriteError
+ rootCause
} else {
wrapAsLogWriteFailure(batchId, resolvedCheckpointRoot, rootCause)
}
@@ -332,16 +359,13 @@ class AsyncProgressTrackingMicroBatchExecution(
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
+ throw getAsyncOperationFailure(
+ thrownError = e,
+ isInterruption = StreamExecution.isInterruptionException(
+ e, sparkSession.sparkContext),
+ interruptingError = Option(interruptingAsyncError.get()),
+ notifierError = errorNotifier.getError()).getOrElse(e)
}
}
@@ -466,6 +490,35 @@ object AsyncProgressTrackingMicroBatchExecution {
val ASYNC_PROGRESS_TRACKING_OVERRIDE_SINK_SUPPORT_CHECK =
"_asyncProgressTrackingOverrideSinkSupportCheck"
+ /**
+ * @return an async-operation wrapper with the interrupting error as its
cause and any
+ * independent earlier error as suppressed context, or None when the
thrown error is
+ * unrelated or fatal
+ */
+ private[streaming] def getAsyncOperationFailure(
+ thrownError: Throwable,
+ isInterruption: Boolean,
+ interruptingError: Option[Throwable],
+ notifierError: Option[Throwable]): Option[SparkRuntimeException] = {
+ val primaryError = interruptingError.orElse(notifierError)
+ val shouldWrap = (isInterruption && primaryError.isDefined) ||
+ (NonFatal(thrownError) && interruptingError.isDefined &&
+ notifierError.exists(_ eq thrownError))
+
+ primaryError.filter(_ => shouldWrap).map { error =>
+ val failure = new SparkRuntimeException(
+ errorClass = "STREAMING_ASYNC_OPERATION_FAILED",
+ messageParameters = Map("message" -> error.getMessage),
+ cause = error)
+ notifierError
+ .filterNot(_ eq error)
+ // A shared node renders as [CIRCULAR REFERENCE] after the first
branch prints it.
+ .filter(ErrorNotifier.haveDisjointThrowableGraphs(error, _))
+ .foreach(failure.addSuppressed)
+ failure
+ }
+ }
+
private def getAsyncProgressTrackingCheckpointingIntervalMs(
extraOptions: Map[String, String],
trigger: Trigger): Long = {
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
index 453072245fb3..fc6bb91bd956 100644
---
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
@@ -18,12 +18,16 @@
package org.apache.spark.sql.execution.streaming
import java.io.File
+import java.util.concurrent.{CompletionException, CountDownLatch,
ExecutionException, TimeUnit}
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.{SparkException, SparkRuntimeException, SparkThrowable}
+import org.apache.spark.errors.SparkCoreErrors
+import org.apache.spark.sql.{Encoder, SparkSession}
+import org.apache.spark.sql.connector.read.streaming.{Offset => OffsetV2,
ReadLimit}
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
@@ -33,13 +37,33 @@ 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
+import org.apache.spark.util.{SparkErrorUtils, Utils}
private class UnsupportedSink extends MemorySink {
override def name(): String = "UnsupportedSink"
}
+private class CategorizedExecutionException(cause: Throwable)
+ extends ExecutionException("categorized execution failure", cause) with
SparkThrowable {
+
+ override def getCondition: String = "TEST_ERROR_CLASS"
+}
+
+private class BlockingLatestOffsetMemoryStream(
+ sparkSession: SparkSession,
+ latestOffsetEntered: CountDownLatch,
+ releaseLatestOffset: CountDownLatch)(
+ implicit encoder: Encoder[Int])
+ extends LowLatencyMemoryStream[Int](Int.MaxValue, sparkSession) {
+
+ override def latestOffset(startOffset: OffsetV2, limit: ReadLimit): OffsetV2
= {
+ latestOffsetEntered.countDown()
+ releaseLatestOffset.await()
+ super.latestOffset(startOffset, limit)
+ }
+}
+
class AsyncProgressTrackingRealTimeModeSuite
extends StreamRealTimeModeManualClockSuiteBase with BeforeAndAfter with
Matchers {
@@ -340,19 +364,11 @@ class AsyncProgressTrackingRealTimeModeSuite
AddData(inputData, 10 until 20: _*),
CheckAnswerWithTimeout(60000, 0 until 20: _*),
advanceRealTimeClock,
- ExpectFailure[Exception](
+ ExpectFailure[SparkRuntimeException](
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]
+ val asyncFailure = e.asInstanceOf[SparkRuntimeException]
+ assert(asyncFailure.getCondition ===
"STREAMING_ASYNC_OPERATION_FAILED")
+ val sparkEx =
asyncFailure.getCause.asInstanceOf[SparkException]
checkError(
sparkEx,
expectedErrorClass,
@@ -361,8 +377,7 @@ class AsyncProgressTrackingRealTimeModeSuite
// 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)
@@ -371,6 +386,77 @@ class AsyncProgressTrackingRealTimeModeSuite
}
}
+ private def testInjectedAsyncFailure(
+ asyncWriteError: Throwable,
+ retainedError: Option[Throwable] = None,
+ batchId: Long = 7L)(
+ assertFailure: SparkRuntimeException => Unit): Unit = {
+ withTempDir { checkpointDir =>
+ val latestOffsetEntered = new CountDownLatch(1)
+ val releaseLatestOffset = new CountDownLatch(1)
+ val inputData = new BlockingLatestOffsetMemoryStream(
+ spark, latestOffsetEntered, releaseLatestOffset)
+
+ testStream(
+ inputData.toDF(),
+ outputMode = OutputMode.Update(),
+ sink = new ContinuousMemorySink(),
+ extraOptions = Map(
+ ASYNC_PROGRESS_TRACKING_ENABLED -> "true",
+ ASYNC_PROGRESS_TRACKING_CHECKPOINTING_INTERVAL_MS -> "0"))(
+ StartStream(checkpointLocation = checkpointDir.getAbsolutePath),
+ Execute { query =>
+ try {
+ assert(latestOffsetEntered.await(60, TimeUnit.SECONDS))
+ val execution =
query.asInstanceOf[AsyncProgressTrackingMicroBatchExecution]
+ retainedError.foreach(execution.errorNotifier.markError)
+ execution.handleAsyncLogWriteError(
+ asyncWriteError, batchId, StreamingErrors.offsetLogWriteFailure)
+ } finally {
+ releaseLatestOffset.countDown()
+ }
+ },
+ ExpectFailure[SparkRuntimeException] { error =>
+ assertFailure(error.asInstanceOf[SparkRuntimeException])
+ })
+ }
+ }
+
+ private def testRetainedAndInterruptingErrors(
+ retainedError: Throwable,
+ writeError: Throwable,
+ expectedSuppressedErrors: Seq[Throwable]): Unit = {
+ testInjectedAsyncFailure(writeError, retainedError = Some(retainedError))
{ failure =>
+ assert(failure.getCondition === "STREAMING_ASYNC_OPERATION_FAILED")
+ val categorized = failure.getCause.asInstanceOf[SparkException]
+ checkError(
+ categorized,
+ StreamingErrors.OFFSET_LOG_WRITE_FAILURE,
+ parameters = Map("batchId" -> "7", "checkpointLocation" -> ".*"),
+ matchPVals = true)
+ assert(categorized.getCause eq writeError)
+ assert(failure.getSuppressed.toSeq === expectedSuppressedErrors)
+ assert(retainedError.getSuppressed.isEmpty)
+ assert(!SparkErrorUtils.stackTraceToString(failure).contains("CIRCULAR
REFERENCE"))
+ }
+ }
+
+ private def testWrappedCategorizedFailure(
+ wrap: Throwable => Throwable,
+ preserveWrapper: Boolean = false): Unit = {
+ val categorizedError = StreamingErrors.offsetLogWriteFailure(
+ batchId = 1L,
+ checkpointLocation = "/checkpoint",
+ cause = new java.io.IOException("checkpoint authorization failure"))
+ val asyncWriteError = wrap(categorizedError)
+ val expectedCause = if (preserveWrapper) asyncWriteError else
categorizedError
+
+ testInjectedAsyncFailure(asyncWriteError, batchId = 1L) { failure =>
+ assert(failure.getCondition === "STREAMING_ASYNC_OPERATION_FAILED")
+ assert(failure.getCause eq expectedCause)
+ }
+ }
+
test("throw exceptions immediately on commit log write failure") {
testAsyncLogWriteFailure("commits",
StreamingErrors.COMMIT_LOG_WRITE_FAILURE)
}
@@ -378,4 +464,64 @@ class AsyncProgressTrackingRealTimeModeSuite
test("throw exceptions immediately on offset log write failure") {
testAsyncLogWriteFailure("offsets",
StreamingErrors.OFFSET_LOG_WRITE_FAILURE)
}
+
+ test("report the async error that interrupted RTM instead of an earlier
retained error") {
+ val retainedError = new java.io.IOException("earlier async purge failure")
+ testRetainedAndInterruptingErrors(
+ retainedError = retainedError,
+ writeError = new java.io.IOException("current offset write failure"),
+ expectedSuppressedErrors = Seq(retainedError))
+ }
+
+ test("wrap a non-fatal notifier error observed before the RTM interruption")
{
+ val retainedError = new java.io.IOException("earlier async failure")
+ val interruptingError = new java.io.IOException("interrupting async
failure")
+ val failure =
AsyncProgressTrackingMicroBatchExecution.getAsyncOperationFailure(
+ thrownError = retainedError,
+ isInterruption = false,
+ interruptingError = Some(interruptingError),
+ notifierError = Some(retainedError)).get
+
+ assert(failure.getCause eq interruptingError)
+ assert(failure.getSuppressed.toSeq === Seq(retainedError))
+ }
+
+ test("do not wrap a fatal notifier error observed before the RTM
interruption") {
+ val fatalError = SparkCoreErrors.outOfMemoryError(
+ requestedBytes = 1L,
+ receivedBytes = 0L,
+ consumerBreakdown = "test")
+ val failure =
AsyncProgressTrackingMicroBatchExecution.getAsyncOperationFailure(
+ thrownError = fatalError,
+ isInterruption = false,
+ interruptingError = Some(new RuntimeException("interrupting async
failure")),
+ notifierError = Some(fatalError))
+
+ assert(failure.isEmpty)
+ }
+
+ test("do not suppress a retained error already in the interrupting error
graph") {
+ val retainedError = new java.io.IOException("earlier async failure")
+ testRetainedAndInterruptingErrors(
+ retainedError = retainedError,
+ writeError = retainedError,
+ expectedSuppressedErrors = Seq.empty)
+ }
+
+ Seq[(String, Throwable => Throwable)](
+ "CompletionException" -> ((error: Throwable) => new
CompletionException(error)),
+ "ExecutionException" -> ((error: Throwable) => new
ExecutionException(error)),
+ "mixed future wrappers" -> ((error: Throwable) =>
+ new CompletionException(new ExecutionException(error)))).foreach {
+ case (wrapperName, wrap) =>
+ test(s"unwrap a categorized failure nested in $wrapperName") {
+ testWrappedCategorizedFailure(wrap)
+ }
+ }
+
+ test("preserve a categorized failure that is also an ExecutionException") {
+ testWrappedCategorizedFailure(
+ error => new CategorizedExecutionException(error),
+ preserveWrapper = true)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
