stop the query rather than retry looping when task tries to retry

Author: jose-torres

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala

@@ -89,15 +89,7 @@ class ContinuousExecution(
     }
 
     do {
-      try {
-        runContinuous(sparkSessionForStream)
-      } catch {
-        // Capture task retries (transformed to an exception by ContinuousDataSourceRDDIter) and
-        // convert them to global retries by letting the while loop spin.
-        case s: SparkException
-            if s.getCause != null && s.getCause.getCause != null &&
-                s.getCause.getCause.isInstanceOf[ContinuousTaskRetryException] => ()
-      }
+      runContinuous(sparkSessionForStream)
     } while (state.updateAndGet(stateUpdate) == ACTIVE)
   }
 

sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala

@@ -201,7 +201,7 @@ class ContinuousSuite extends ContinuousSuiteBase {
       StopStream)
   }
 
-  test("task failure triggers a ContinuousExecution restart") {
+  test("task failure kills the query") {
     val df = spark.readStream
       .format("rate")
       .option("numPartitions", "5")
@@ -218,23 +218,21 @@ class ContinuousSuite extends ContinuousSuiteBase {
     }
     spark.sparkContext.addSparkListener(listener)
 
-
-    var originalRunId: UUID = null
     testStream(df, useV2Sink = true)(
       StartStream(Trigger.Continuous(100)),
       Execute(waitForRateSourceTriggers(_, 2)),
       Execute { query =>
         // Wait until a task is started, then kill its first attempt.
         eventually(timeout(streamingTimeout)) { assert(taskId != -1) }
-        originalRunId = query.runId
         spark.sparkContext.killTaskAttempt(taskId)
-      },
-      Execute(waitForRateSourceTriggers(_, 4)),
-      IncrementEpoch(),
-      // Rather than just restarting the task we killed, there should have been a
-      // ContinuousExecution restart changing the run ID.
-      AssertOnQuery(_.runId != originalRunId),
-      CheckAnswerRowsContains(scala.Range(0, 20).map(Row(_))))
+        eventually(timeout(streamingTimeout)) {
+          assert(query.exception.isDefined)
+        }
+        assert(
+          query.exception.get.getCause != null &&
+          query.exception.get.getCause.getCause != null &&
+          query.exception.get.getCause.getCause.getCause.isInstanceOf[ContinuousTaskRetryException])
+      })
 
     spark.sparkContext.removeSparkListener(listener)
   }