You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/12 16:40:39 UTC
(spark) branch branch-4.x updated: [SPARK-58698][SS][TEST] Add additional tests for RTM
This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 83fad5aebd27 [SPARK-58698][SS][TEST] Add additional tests for RTM
83fad5aebd27 is described below
commit 83fad5aebd27f73ba490e8952eebf0a312bb3cdd
Author: Boyang Jerry Peng <je...@databricks.com>
AuthorDate: Thu Aug 13 00:40:05 2026 +0800
[SPARK-58698][SS][TEST] Add additional tests for RTM
### What changes were proposed in this pull request?
This is a test-only PR. It adds end-to-end test coverage for Real-Time Mode (RTM) over a Kafka source, for windowed aggregation over Kafka, and for RTM queries coexisting with other
queries in the same cluster. No production code is changed.
The following suites are added (or extended):
- **`KafkaRealTimeModeBaseSuite`** (new, abstract) - a shared harness for the Kafka RTM suites. It starts an embedded Kafka cluster, configures an isolated `SparkSession` with the RTM
trigger and a manual clock, and provides helpers to write a query's output to Kafka and read a topic back, so each concrete suite only expresses the query under test.
- **`KafkaRealTimeModeSuite`** (new/extended) - core RTM behavior over a Kafka source (simple map, adding a partition mid-stream, re-fetching the latest offset at batch end, unions, the
self-union case, a Kafka + non-Kafka union, aggregation, and consumer-pool reuse).
- **`KafkaRealTimeModeAggregationSuite`** (new) - windowed aggregation over a Kafka source across the aggregate functions `max`, `min`, `sum`, and `avg`.
- **`KafkaRealTimeModeWindowSuite`** (new) - tumbling- and sliding-window count aggregations over a Kafka source in RTM.
- **`StreamRealTimeModeCoexistenceSuite`** (new) - RTM and other queries running at the same time in one cluster: an RTM and a micro-batch multi-stage query concurrently; a batch query
with a shuffle running while an RTM query is active; and two concurrent RTM queries, each with its own pipelined group, verified to be co-scheduled via a `SparkListener` that observes a
stage of each query running at the same time.
### Why are the changes needed?
RTM support was built up across the pipelined-shuffle / RTM PR stack, but the Kafka source - the most common streaming source - and the "RTM alongside other queries" scenario were not
directly covered by end-to-end tests. These suites exercise RTM against a real Kafka source (including windowed aggregation) and confirm that RTM and other queries share a cluster without
interfering, guarding against regressions in the source integration, windowed aggregation, and the scheduler/shuffle coexistence paths.
### Does this PR introduce _any_ user-facing change?
No. Tests only.
### How was this patch tested?
This PR is the tests. All added suites pass locally:
- `KafkaRealTimeModeSuite`, `KafkaRealTimeModeAggregationSuite`, and `KafkaRealTimeModeWindowSuite` (embedded Kafka broker).
- `StreamRealTimeModeCoexistenceSuite` (3 tests).
- `KafkaRealTimeModeBaseSuite` is abstract and has no tests of its own.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Code
Closes #57903 from jerrypeng/stack/pipelined-shuffle-pr10-kafka-rtm-tests.
Authored-by: Boyang Jerry Peng <je...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 0bf401810d997a7988863c8b2c78f318277da392)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../KafkaRealTimeModeAggregationSuite.scala | 271 +++++++++++++++++++++
.../sql/kafka010/KafkaRealTimeModeBaseSuite.scala | 184 ++++++++++++++
.../sql/kafka010/KafkaRealTimeModeSuite.scala | 80 ++++++
.../kafka010/KafkaRealTimeModeWindowSuite.scala | 247 +++++++++++++++++++
.../StreamRealTimeModeCoexistenceSuite.scala | 255 +++++++++++++++++++
5 files changed, 1037 insertions(+)
diff --git a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeAggregationSuite.scala b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeAggregationSuite.scala
new file mode 100644
index 000000000000..28b397f13e44
--- /dev/null
+++ b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeAggregationSuite.scala
@@ -0,0 +1,271 @@
+/*
+ * 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.kafka010
+
+import scala.collection.mutable
+
+import org.scalatest.time.SpanSugar._
+
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+import org.apache.spark.sql.execution.streaming.runtime.StreamingQueryWrapper
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.{StringType, StructField, StructType}
+
+class KafkaRealTimeModeAggregationSuite extends KafkaRealTimeModeBaseSuite {
+
+ test("tumbling window max") {
+ runTest {
+ case params @ TestParams(query, clock, read, outputTopic, checkpointDir) =>
+ val tumblingWindowDuration = 10
+ val numRows = 10
+ val readPart = read
+ .toDF()
+ .select(col("_1").as("timestamp").cast("TIMESTAMP"), col("_2").as("value"))
+ .groupBy(window(column("timestamp"), s"${tumblingWindowDuration} seconds"))
+ .max()
+ .select(
+ concat(
+ col("window").cast("STRING"),
+ lit("-"),
+ col("max(value)").cast("STRING")
+ ).as("value")
+ )
+
+ params.query =
+ writeToKafka("tumbling_window_max_low_latency", outputTopic, checkpointDir, readPart)
+
+ val expectedResults = mutable.ListBuffer[GenericRowWithSchema]()
+
+ for (i <- 0 until 3) {
+ for (k <- (1 to numRows).reverse) {
+ val data = ((i * 10).toLong, k)
+ read.addData(0, Seq(data))
+
+ val windowDurationMs = tumblingWindowDuration * 1000
+
+ val startTime = getDateTimeString(((i + 1) * windowDurationMs) - windowDurationMs)
+ val endTime = getDateTimeString((i + 1) * windowDurationMs)
+
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{$startTime, $endTime}-${numRows}"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+
+ eventually(timeout(60.seconds)) {
+ checkAnswer(readKafkaTopic(outputTopic), expectedResults.toSeq)
+ }
+ // advance to next batch
+ clock.advance(1000)
+
+ eventually(timeout(60.seconds)) {
+ params.query
+ .asInstanceOf[StreamingQueryWrapper]
+ .streamingQuery
+ .getLatestExecutionContext()
+ .batchId should be(i + 1)
+ params.query.lastProgress.sources(0).numInputRows should be(numRows)
+ }
+ }
+ }
+ }
+
+ test("tumbling window min") {
+ runTest {
+ case params @ TestParams(query, clock, read, outputTopic, checkpointDir) =>
+ val tumblingWindowDuration = 10
+ val numRows = 10
+
+ val readPart = read
+ .toDF()
+ .select(col("_1").as("timestamp").cast("TIMESTAMP"), col("_2").as("value"))
+ .groupBy(window(column("timestamp"), s"${tumblingWindowDuration} seconds"))
+ .min()
+ .select(
+ concat(
+ col("window").cast("STRING"),
+ lit("-"),
+ col("min(value)").cast("STRING")
+ ).as("value")
+ )
+
+ params.query =
+ writeToKafka("tumbling_window_min_low_latency", outputTopic, checkpointDir, readPart)
+
+ val expectedResults = mutable.ListBuffer[GenericRowWithSchema]()
+
+ for (i <- 0 until 3) {
+ for (k <- (1 to numRows)) {
+ val data = ((i * 10).toLong, k)
+ read.addData(0, Seq(data))
+
+ val windowDurationMs = tumblingWindowDuration * 1000
+
+ val startTime = getDateTimeString(((i + 1) * windowDurationMs) - windowDurationMs)
+ val endTime = getDateTimeString((i + 1) * windowDurationMs)
+
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{$startTime, $endTime}-${1}"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+
+ eventually(timeout(60.seconds)) {
+ checkAnswer(readKafkaTopic(outputTopic), expectedResults.toSeq)
+ }
+ // advance to next batch
+ clock.advance(1000)
+
+ eventually(timeout(60.seconds)) {
+ params.query
+ .asInstanceOf[StreamingQueryWrapper]
+ .streamingQuery
+ .getLatestExecutionContext()
+ .batchId should be(i + 1)
+ params.query.lastProgress.sources(0).numInputRows should be(numRows)
+ }
+ }
+ }
+ }
+
+ test("tumbling window sum") {
+ runTest {
+ case params @ TestParams(query, clock, read, outputTopic, checkpointDir) =>
+ val tumblingWindowDuration = 10
+ val numRows = 10
+
+ val readPart = read
+ .toDF()
+ .select(col("_1").as("timestamp").cast("TIMESTAMP"), col("_2").as("value"))
+ .groupBy(window(column("timestamp"), s"${tumblingWindowDuration} seconds"))
+ .sum()
+ .select(
+ concat(
+ col("window").cast("STRING"),
+ lit("-"),
+ col("sum(value)").cast("STRING")
+ ).as("value")
+ )
+
+ params.query =
+ writeToKafka("tumbling_window_sum_low_latency", outputTopic, checkpointDir, readPart)
+
+ val expectedResults = mutable.ListBuffer[GenericRowWithSchema]()
+
+ for (i <- 0 until 3) {
+ var sum = 0
+ for (k <- (1 to numRows)) {
+ val data = ((i * 10).toLong, k)
+ read.addData(0, Seq(data))
+
+ val windowDurationMs = tumblingWindowDuration * 1000
+
+ val startTime = getDateTimeString(((i + 1) * windowDurationMs) - windowDurationMs)
+ val endTime = getDateTimeString((i + 1) * windowDurationMs)
+
+ sum += k
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{$startTime, $endTime}-${sum}"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+
+ eventually(timeout(60.seconds)) {
+ checkAnswer(readKafkaTopic(outputTopic), expectedResults.toSeq)
+ }
+ // advance to next batch
+ clock.advance(1000)
+
+ eventually(timeout(60.seconds)) {
+ params.query
+ .asInstanceOf[StreamingQueryWrapper]
+ .streamingQuery
+ .getLatestExecutionContext()
+ .batchId should be(i + 1)
+ params.query.lastProgress.sources(0).numInputRows should be(numRows)
+ }
+ }
+ }
+ }
+
+ test("tumbling window avg") {
+ runTest {
+ case params @ TestParams(query, clock, read, outputTopic, checkpointDir) =>
+ val tumblingWindowDuration = 10
+ val numRows = 10
+
+ val readPart = read
+ .toDF()
+ .select(col("_1").as("timestamp").cast("TIMESTAMP"), col("_2").as("value"))
+ .groupBy(window(column("timestamp"), s"${tumblingWindowDuration} seconds"))
+ .avg()
+ .select(
+ concat(
+ col("window").cast("STRING"),
+ lit("-"),
+ col("avg(value)").cast("INT").cast("STRING")
+ ).as("value")
+ )
+
+ params.query =
+ writeToKafka("tumbling_window_avg_low_latency", outputTopic, checkpointDir, readPart)
+
+ val expectedResults = mutable.ListBuffer[GenericRowWithSchema]()
+
+ for (i <- 0 until 3) {
+ var sum = 0
+ for (k <- (1 to numRows)) {
+ // Feed varying values (the row index k) so the assertion below actually exercises the
+ // average rather than passing for any single-value aggregate: a constant input would
+ // make avg == first == last == that value.
+ val data = ((i * 10).toLong, k)
+ read.addData(0, Seq(data))
+
+ val windowDurationMs = tumblingWindowDuration * 1000
+
+ val startTime = getDateTimeString(((i + 1) * windowDurationMs) - windowDurationMs)
+ val endTime = getDateTimeString((i + 1) * windowDurationMs)
+
+ // Update mode emits the running average after each record: sum(1..k) / k. Cast to INT
+ // to match the query's avg(value) cast.
+ sum += k
+ val runningAvg = sum / k
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{$startTime, $endTime}-$runningAvg"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+
+ eventually(timeout(60.seconds)) {
+ checkAnswer(readKafkaTopic(outputTopic), expectedResults.toSeq)
+ }
+ // advance to next batch
+ clock.advance(1000)
+
+ eventually(timeout(60.seconds)) {
+ params.query
+ .asInstanceOf[StreamingQueryWrapper]
+ .streamingQuery
+ .getLatestExecutionContext()
+ .batchId should be(i + 1)
+ params.query.lastProgress.sources(0).numInputRows should be(numRows)
+ }
+ }
+ }
+ }
+}
diff --git a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeBaseSuite.scala b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeBaseSuite.scala
new file mode 100644
index 000000000000..8eefa70682da
--- /dev/null
+++ b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeBaseSuite.scala
@@ -0,0 +1,184 @@
+/*
+ * 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.kafka010
+
+import java.io.File
+import java.time.{Instant, ZoneId}
+import java.time.format.DateTimeFormatter
+
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.time.SpanSugar._
+
+import org.apache.spark.{SparkContext, ThreadAudit}
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.execution.datasources.v2.LowLatencyClock
+import org.apache.spark.sql.execution.streaming.RealTimeTrigger
+import org.apache.spark.sql.execution.streaming.sources.LowLatencyMemoryStream
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery}
+import org.apache.spark.sql.streaming.util.{GlobalSingletonManualClock, StreamManualClock}
+import org.apache.spark.sql.test.TestSparkSession
+import org.apache.spark.util.Utils
+
+abstract class KafkaRealTimeModeBaseSuite
+ extends KafkaSourceTest
+ with ThreadAudit
+ with BeforeAndAfterEach
+ with Matchers {
+
+ import testImplicits._
+
+ private def defaultTriggerBatchDurationMs: Long = 1000L
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ // testing to make sure the cluster is usable
+ testUtils.createTopic("_test")
+ testUtils.sendMessage(new ProducerRecord[String, String]("_test", "", ""))
+ testUtils.deleteTopic("_test")
+ logInfo("Kafka cluster setup complete....")
+
+ spark.conf.set(SQLConf.SHUFFLE_PARTITIONS.key, 5)
+ spark.conf.set(
+ SQLConf.STATE_STORE_PROVIDER_CLASS.key,
+ "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider"
+ )
+ spark.conf.set("spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled", "true")
+ spark.conf.set("spark.sql.streaming.stateStore.rocksdb.trackTotalNumberOfRows", "false")
+ spark.conf.set("spark.sql.streaming.stateStore.checkpointFormatVersion", "2")
+ spark.conf.set(
+ SQLConf.STREAMING_REAL_TIME_MODE_MIN_BATCH_DURATION,
+ defaultTriggerBatchDurationMs
+ )
+ }
+
+ override protected def createSparkSession =
+ new TestSparkSession(
+ new SparkContext(
+ // Ensure we have enough for both stages. 5 source partitions and 5 shuffle partitions
+ "local[15]",
+ "microbatch-context",
+ sparkConf
+ .set("spark.sql.testkey", "true")
+ .set("spark.sql.shuffle.partitions", "5")
+ .set("spark.sql.adaptive.enabled", "false")
+ .set(
+ "spark.executor.extraJavaOptions",
+ "-Dio.netty.leakDetection.level=paranoid"
+ )
+ )
+ )
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ GlobalSingletonManualClock.reset()
+ }
+
+ protected def writeToKafka(
+ queryName: String,
+ outputTopic: String,
+ checkpointDir: File,
+ df: DataFrame): StreamingQuery = {
+ df.writeStream
+ .outputMode(OutputMode.Update())
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("topic", outputTopic)
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .queryName(queryName)
+ // The batch duration set here doesn't matter because we manually control batch durations
+ // via the manual clock.
+ .trigger(RealTimeTrigger(defaultTriggerBatchDurationMs))
+ .start()
+ }
+
+ protected def readKafkaTopic(topic: String): DataFrame = {
+ spark.read
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+ .select(col("value").cast("STRING"))
+ }
+
+
+ protected def runTest(test: (TestParams) => Unit): Unit = {
+ withTempDir { checkpointDir =>
+ val outputTopic = newTopic()
+ testUtils.createTopic(outputTopic, partitions = 5)
+
+ val clock = new GlobalSingletonManualClock()
+
+ LowLatencyClock.setClock(clock)
+ val read = LowLatencyMemoryStream[(Long, Int)](5)
+ val param = TestParams(null, clock, read, outputTopic, checkpointDir)
+ try {
+ test(param)
+ } finally {
+ if (param.query != null) {
+ param.query.stop()
+ }
+
+ try {
+ eventually(timeout(60.seconds)) {
+
+ val currentRunningTasks = sparkContext.statusTracker.getExecutorInfos
+ .map(
+ i =>
+ s"[host: ${i.host()}" +
+ s" port: ${i.port} tasks:${i.numRunningTasks()}]"
+ )
+ .toList
+
+ logInfo(s"Current tasks: ${currentRunningTasks}")
+
+ assert(
+ spark.sparkContext.statusTracker.getExecutorInfos.map(_.numRunningTasks()).sum <= 0,
+ currentRunningTasks
+ )
+ }
+ } catch {
+ case t: Throwable =>
+ // Best-effort diagnostic for a task that never wound down.
+ logWarning(s"Tasks still running after the query stopped", t)
+ logWarning(Utils.getThreadDump().map(_.toString).mkString("\n"))
+ throw t
+ }
+ }
+ }
+ }
+
+ protected def getDateTimeString(millis: Long): String = {
+ val instant = Instant.ofEpochMilli(millis)
+ val formatter = DateTimeFormatter
+ .ofPattern("yyyy-MM-dd HH:mm:ss")
+ .withZone(ZoneId.systemDefault())
+ formatter.format(instant)
+ }
+}
+
+case class TestParams(
+ var query: StreamingQuery,
+ clock: StreamManualClock,
+ read: LowLatencyMemoryStream[(Long, Int)],
+ outputTopic: String,
+ checkpointDir: File)
diff --git a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeSuite.scala b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeSuite.scala
index ae23b53fc35a..2336bcdbeba4 100644
--- a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeSuite.scala
+++ b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeSuite.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.execution.datasources.v2.LowLatencyClock
import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.execution.streaming.sources.{ContinuousMemorySink, LowLatencyMemoryStream}
import org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider
+import org.apache.spark.sql.functions.{count, timestamp_seconds, window}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
@@ -679,6 +680,85 @@ class KafkaRealTimeModeSuite
}
)
}
+
+ // Aggregation over a Kafka source in Real-Time Mode. The aggregate is planned as the streamline
+ // operator, which merges each record against state and emits as it goes, so in update mode a key
+ // seen twice produces two outputs -- the running value after each record.
+ test("aggregation over a Kafka source") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 2)
+
+ testUtils.sendMessages(topic, Array("a", "b"), Some(0))
+ testUtils.sendMessages(topic, Array("a"), Some(1))
+
+ val aggregated = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+ .selectExpr("CAST(value AS STRING) AS key")
+ .groupBy($"key")
+ .agg(count("*").as("count"))
+ .as[(String, Long)]
+
+ testStream(aggregated, Update, sink = new ContinuousMemorySink())(
+ StartStream(),
+ // "a" arrives twice, so it is emitted at count 1 and again at count 2.
+ CheckAnswerWithTimeout(60000, ("a", 1L), ("a", 2L), ("b", 1L)),
+ WaitUntilCurrentBatchProcessed,
+ new ExternalAction() {
+ override def runAction(): Unit = {
+ testUtils.sendMessages(topic, Array("b", "c"), Some(0))
+ }
+ },
+ CheckAnswerWithTimeout(30000,
+ ("a", 1L), ("a", 2L), ("b", 1L), ("b", 2L), ("c", 1L)),
+ WaitUntilCurrentBatchProcessed,
+ StopStream,
+ new ExternalAction() {
+ override def runAction(): Unit = {
+ testUtils.sendMessages(topic, Array("a"), Some(1))
+ }
+ },
+ StartStream(),
+ // The counts continue from the committed state across the restart rather than restarting.
+ CheckAnswerWithTimeout(30000,
+ ("a", 1L), ("a", 2L), ("b", 1L), ("b", 2L), ("c", 1L), ("a", 3L)),
+ WaitUntilCurrentBatchProcessed)
+ }
+
+ // A tumbling window aggregation over a Kafka source, where the grouping key is derived from an
+ // event time column rather than being a bare field.
+ test("tumbling window aggregation over a Kafka source") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 2)
+
+ // Values are seconds since the epoch; a 10 second window buckets 1-9 together and 11-19 next.
+ testUtils.sendMessages(topic, Array("1", "2"), Some(0))
+ testUtils.sendMessages(topic, Array("11"), Some(1))
+
+ val windowed = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+ .selectExpr("CAST(value AS STRING) AS value")
+ .select(timestamp_seconds($"value".cast("long")).as("eventTime"))
+ .groupBy(window($"eventTime", "10 seconds").as("window"))
+ .agg(count("*").as("count"))
+ .select($"window".getField("end").cast("long").as[Long], $"count".as[Long])
+
+ testStream(windowed, Update, sink = new ContinuousMemorySink())(
+ StartStream(),
+ // window ending at 10 holds 1 and 2, emitted at count 1 then 2; window ending at 20 holds 11.
+ CheckAnswerWithTimeout(60000, (10L, 1L), (10L, 2L), (20L, 1L)),
+ WaitUntilCurrentBatchProcessed)
+ }
+
}
class KafkaConsumerPoolRealTimeModeSuite
diff --git a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeWindowSuite.scala b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeWindowSuite.scala
new file mode 100644
index 000000000000..98d3077a9784
--- /dev/null
+++ b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRealTimeModeWindowSuite.scala
@@ -0,0 +1,247 @@
+/*
+ * 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.kafka010
+
+import scala.collection.mutable
+
+import org.scalatest.time.SpanSugar._
+
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+import org.apache.spark.sql.execution.streaming.runtime.StreamingQueryWrapper
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.{StringType, StructField, StructType}
+
+class KafkaRealTimeModeWindowSuite extends KafkaRealTimeModeBaseSuite {
+
+ test("tumbling window count") {
+ runTest {
+ case params @ TestParams(query, clock, read, outputTopic, checkpointDir) =>
+ val tumblingWindowDuration = 10
+ val numRows = 10
+
+ val readPart = read
+ .toDF()
+ .select(col("_1").as("timestamp").cast("TIMESTAMP"), col("_2").as("value"))
+ .groupBy(
+ window(column("timestamp"), s"${tumblingWindowDuration} seconds"),
+ column("value")
+ )
+ .count()
+ .select(
+ concat(
+ col("window").cast("STRING"),
+ lit("-"),
+ col("value").cast("STRING"),
+ lit("-"),
+ col("count").cast("STRING")
+ ).as("value")
+ )
+
+ params.query = writeToKafka("tumbling_window_count_low_latency",
+ outputTopic, checkpointDir, readPart)
+
+ val expectedResults = mutable.ListBuffer[GenericRowWithSchema]()
+
+ for (i <- 0 until 3) {
+ for (k <- 0 until numRows) {
+ val value = k % 2
+ val data = ((i * 10).toLong, value)
+ read.addData({
+ data
+ })
+
+ /**
+ * results should be something like this
+ * {1969-12-31 16:00:00, 1969-12-31 16:00:10}-0-1
+ * {1969-12-31 16:00:00, 1969-12-31 16:00:10}-1-1
+ * {1969-12-31 16:00:00, 1969-12-31 16:00:10}-0-2
+ * {1969-12-31 16:00:00, 1969-12-31 16:00:10}-1-2
+ */
+ val windowDurationMs = tumblingWindowDuration * 1000
+
+ val startTime = getDateTimeString(((i + 1) * windowDurationMs) - windowDurationMs)
+ val endTime = getDateTimeString((i + 1) * windowDurationMs)
+
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{$startTime, $endTime}-$value-${Math.ceil((k + 1) / 2.0).toInt}"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+
+ eventually(timeout(60.seconds)) {
+ checkAnswer(readKafkaTopic(outputTopic), expectedResults.toSeq)
+ }
+ // advance to next batch
+ clock.advance(1000)
+
+ eventually(timeout(60.seconds)) {
+ params.query
+ .asInstanceOf[StreamingQueryWrapper]
+ .streamingQuery
+ .getLatestExecutionContext()
+ .batchId should be(i + 1)
+ params.query.lastProgress.sources(0).numInputRows should be(numRows)
+ }
+ }
+ }
+ }
+
+ test("sliding window count") {
+ runTest {
+ case params @ TestParams(query, clock, read, outputTopic, checkpointDir) =>
+ val numRows = 10
+ val slideWindowDuration = 5
+ val windowDuration = 10
+
+ val readPart = read
+ .toDF()
+ .select(col("_1").as("timestamp").cast("TIMESTAMP"), col("_2").as("value"))
+ .groupBy(
+ window(
+ column("timestamp"),
+ s"$windowDuration seconds",
+ s"$slideWindowDuration seconds"
+ ),
+ column("value")
+ )
+ .count()
+ .select(
+ concat(
+ col("window").cast("STRING"),
+ lit("-"),
+ col("value").cast("STRING"),
+ lit("-"),
+ col("count").cast("STRING")
+ ).as("value")
+ )
+
+ params.query =
+ writeToKafka("sliding_window_count_low_latency", outputTopic, checkpointDir, readPart)
+
+ // -5 -> 5
+ val bucket0 = mutable.HashMap[Int, Int]()
+ // 0 -> 10
+ val bucket1 = mutable.HashMap[Int, Int]()
+ // 5 -> 15
+ val bucket2 = mutable.HashMap[Int, Int]()
+ // 10 -> 20
+ val bucket3 = mutable.HashMap[Int, Int]()
+
+ val expectedResults = mutable.ListBuffer[GenericRowWithSchema]()
+ for (i <- 0 until 3) {
+ for (k <- 0 until numRows) {
+ val value = k % 2
+ val data = ((i * 5).toLong, value)
+ read.addData({
+ data
+ })
+
+ /**
+ * Results should be something like
+ *
+ * {1969-12-31 16:00:00, 1969-12-31 16:00:10}-0-1
+ * {1969-12-31 15:59:55, 1969-12-31 16:00:05}-0-1
+ * {1969-12-31 16:00:00, 1969-12-31 16:00:10}-1-1
+ * {1969-12-31 15:59:55, 1969-12-31 16:00:05}-1-1
+ * {1969-12-31 16:00:00, 1969-12-31 16:00:10}-0-2
+ * {1969-12-31 15:59:55, 1969-12-31 16:00:05}-0-2
+ * {1969-12-31 16:00:00, 1969-12-31 16:00:10}-1-2
+ * {1969-12-31 15:59:55, 1969-12-31 16:00:05}-1-2
+ * ...
+ */
+ val ts = data._1
+ if (ts >= -5 && ts < 5) {
+ bucket0(value) = bucket0.getOrElse(value, 0) + 1
+ }
+
+ if (ts >= 0 && ts < 10) {
+ bucket1(value) = bucket1.getOrElse(value, 0) + 1
+ }
+
+ if (ts >= 5 && ts < 15) {
+ bucket2(value) = bucket2.getOrElse(value, 0) + 1
+ }
+
+ if (ts >= 10 && ts < 20) {
+ bucket3(value) = bucket3.getOrElse(value, 0) + 1
+ }
+ }
+
+ bucket0.foreach(pair => {
+ val k = pair._1
+ val count = pair._2
+ for (i <- 1 to count) {
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{${getDateTimeString(-5000)}, ${getDateTimeString(5000)}}-$k-${i}"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+ })
+
+ bucket1.foreach(pair => {
+ val k = pair._1
+ val count = pair._2
+ for (i <- 1 to count) {
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{${getDateTimeString(0)}, ${getDateTimeString(10000)}}-$k-${i}"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+ })
+
+ bucket2.foreach(pair => {
+ val k = pair._1
+ val count = pair._2
+ for (i <- 1 to count) {
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{${getDateTimeString(5000)}, ${getDateTimeString(15000)}}-$k-${i}"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+ })
+
+ bucket3.foreach(pair => {
+ val k = pair._1
+ val count = pair._2
+ for (i <- 1 to count) {
+ expectedResults += new GenericRowWithSchema(
+ Array(s"{${getDateTimeString(10000)}, ${getDateTimeString(20000)}}-$k-${i}"),
+ schema = new StructType().add(StructField("value", StringType))
+ )
+ }
+ })
+
+ eventually(timeout(60.seconds)) {
+ checkAnswer(readKafkaTopic(outputTopic), expectedResults.toSeq)
+ }
+
+ expectedResults.clear()
+ clock.advance(1000)
+
+ eventually(timeout(60.seconds)) {
+ params.query
+ .asInstanceOf[StreamingQueryWrapper]
+ .streamingQuery
+ .getLatestExecutionContext()
+ .batchId should be(i + 1)
+ params.query.lastProgress.sources(0).numInputRows should be(numRows)
+ }
+ }
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeCoexistenceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeCoexistenceSuite.scala
new file mode 100644
index 000000000000..9b1e8b9516d9
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamRealTimeModeCoexistenceSuite.scala
@@ -0,0 +1,255 @@
+/*
+ * 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.streaming
+
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart, SparkListenerStageCompleted, SparkListenerStageSubmitted}
+import org.apache.spark.sql.{ForeachWriter, Row}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.streaming.runtime.{MemoryStream, StreamExecution,
+ StreamingQueryWrapper}
+import org.apache.spark.sql.execution.streaming.sources.{ContinuousMemorySink,
+ LowLatencyMemoryStream}
+import org.apache.spark.sql.functions.count
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Tests that Real-Time Mode (RTM) and MicroBatch Mode (MBM) multi-stage queries can run in the
+ * SAME cluster -- concurrently, in one SparkContext, sharing one set of executors.
+ *
+ * This is the coexistence property that makes RTM usable without dedicating a cluster to it. It
+ * holds because a shuffle's implementation is chosen by DEPENDENCY TYPE, not by a cluster-wide
+ * setting: `SparkEnv.shuffleManagerFor` routes a `PipelinedShuffleDependency` to the pipelined
+ * manager (`spark.shuffle.manager.incremental`, the streaming shuffle) and every other
+ * `ShuffleDependency` to the blocking manager (`spark.shuffle.manager`, sort shuffle). Both
+ * managers are instantiated in the same JVM and neither query has to know about the other.
+ *
+ * Each test uses MULTI-STAGE queries on both sides, since a single-stage query has no shuffle and
+ * so would not exercise routing at all. The RTM side is verified to be genuinely pipelined (rather
+ * than merely running) by asserting on the `pipelined` flag of its exchanges, and the MBM side is
+ * verified to be genuinely NOT pipelined -- a test that only checked both queries produced answers
+ * would pass even if routing collapsed to a single manager.
+ */
+class StreamRealTimeModeCoexistenceSuite extends StreamRealTimeModeSuiteBase {
+
+ import testImplicits._
+
+ /** Every shuffle exchange in the query's last executed plan, with its `pipelined` flag. */
+ private def exchangePipelinedFlags(q: StreamExecution): Seq[Boolean] =
+ q.lastExecution.executedPlan.collect { case s: ShuffleExchangeExec => s.pipelined }
+
+ /** Asserts the query has at least one shuffle and every one of them is pipelined. */
+ private def assertAllExchangesPipelined(q: StreamExecution): Unit = {
+ val flags = exchangePipelinedFlags(q)
+ assert(flags.nonEmpty, "expected at least one shuffle exchange in the RTM plan")
+ assert(flags.forall(identity),
+ s"expected every RTM exchange to be pipelined, got: ${flags.mkString(", ")}")
+ }
+
+ /** Asserts the query has at least one shuffle and none of them is pipelined. */
+ private def assertNoExchangePipelined(q: StreamExecution): Unit = {
+ val flags = exchangePipelinedFlags(q)
+ assert(flags.nonEmpty, "expected at least one shuffle exchange in the MBM plan")
+ assert(!flags.exists(identity),
+ s"expected no MBM exchange to be pipelined, got: ${flags.mkString(", ")}")
+ }
+
+ test("an RTM and an MBM multi-stage query run concurrently in the same cluster") {
+ // Keep both queries' shuffles small: the RTM query's whole pipelined group is gang-admitted, so
+ // its scan + dedup tasks and the MBM query's tasks must all fit the cluster's slots at once.
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+ val rtmInput = LowLatencyMemoryStream[(String, Int)]
+ val mbmInput = MemoryStream[(String, Int)]
+
+ // Both are multi-stage: a shuffle (repartition by key) feeding a stateful dedup.
+ val rtmQuery = rtmInput.toDF().select($"_1".as("key")).dropDuplicates("key").select($"key")
+ val mbmQuery = mbmInput.toDF().select($"_1".as("key")).dropDuplicates("key").select($"key")
+
+ // Start the MBM query first and leave it running for the whole RTM test.
+ val mbmHandle = mbmQuery.writeStream
+ .format("memory")
+ .queryName("coexistence_mbm")
+ .outputMode(OutputMode.Update)
+ .start()
+
+ try {
+ mbmInput.addData(("a", 1), ("b", 1), ("a", 2))
+ mbmHandle.processAllAvailable()
+ checkAnswer(spark.table("coexistence_mbm"), Seq(Row("a"), Row("b")))
+
+ val mbmExec = mbmHandle.asInstanceOf[StreamingQueryWrapper].streamingQuery
+ assertNoExchangePipelined(mbmExec)
+
+ // With the MBM query still active, run an RTM query in the same context.
+ testStream(rtmQuery, OutputMode.Update, Map.empty, new ContinuousMemorySink())(
+ AddData(rtmInput, ("x", 1), ("y", 1), ("x", 2)),
+ StartStream(),
+ CheckAnswerWithTimeout(60000, "x", "y"),
+ Execute { q =>
+ assertAllExchangesPipelined(q)
+ assert(mbmHandle.isActive, "the MBM query must still be running alongside RTM")
+ },
+ StopStream
+ )
+
+ // The MBM query must still make progress AFTER the RTM query has come and gone, proving the
+ // pipelined shuffle did not disturb the blocking manager's state.
+ mbmInput.addData(("c", 1), ("a", 3))
+ mbmHandle.processAllAvailable()
+ checkAnswer(spark.table("coexistence_mbm"), Seq(Row("a"), Row("b"), Row("c")))
+ assertNoExchangePipelined(
+ mbmHandle.asInstanceOf[StreamingQueryWrapper].streamingQuery)
+ } finally {
+ mbmHandle.stop()
+ spark.sql("DROP TABLE IF EXISTS coexistence_mbm")
+ }
+ }
+ }
+
+ test("a batch query with a shuffle runs while an RTM query is active") {
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+ val rtmInput = LowLatencyMemoryStream[(String, Int)]
+ val rtmQuery = rtmInput.toDF().select($"_1".as("key")).dropDuplicates("key").select($"key")
+
+ // A multi-stage BATCH query: groupBy forces a blocking shuffle. Run it mid-RTM-batch.
+ val batchResult = new AtomicReference[Seq[Row]](null)
+
+ testStream(rtmQuery, OutputMode.Update, Map.empty, new ContinuousMemorySink())(
+ AddData(rtmInput, ("x", 1), ("y", 1), ("x", 2)),
+ StartStream(),
+ CheckAnswerWithTimeout(60000, "x", "y"),
+ Execute { q =>
+ assertAllExchangesPipelined(q)
+ // While the RTM batch is still open, a regular batch job with its own shuffle must run to
+ // completion on the same executors, using the blocking shuffle manager.
+ val df = spark.range(0, 100).selectExpr("id % 5 AS k").groupBy("k").agg(count("*"))
+ batchResult.set(df.orderBy("k").collect().toSeq)
+ },
+ Execute { _ =>
+ val rows = batchResult.get()
+ assert(rows != null, "the batch query did not run")
+ assert(rows.length == 5, s"expected 5 groups, got ${rows.length}")
+ assert(rows.forall(_.getLong(1) == 20L), s"expected 20 rows per group, got $rows")
+ },
+ StopStream
+ )
+ }
+ }
+
+ test("two RTM queries run concurrently, each with its own pipelined group") {
+ // Two independent pipelined groups must be admitted and co-scheduled at the same time. Keep the
+ // partition counts low so both groups' gang demands fit the cluster together.
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+ val inputA = LowLatencyMemoryStream[(String, Int)]
+ val inputB = LowLatencyMemoryStream[(String, Int)]
+
+ val queryA = inputA.toDF().select($"_1".as("key")).dropDuplicates("key").select($"key")
+ val queryB = inputB.toDF().select($"_1".as("key")).dropDuplicates("key").select($"key")
+
+ // Track, from the driver, whether a stage of query A and a stage of query B were ever RUNNING
+ // at the same time. Each query's long-running RTM stages are attributed to its query id (via
+ // the job property StreamExecution tags), and `sawBothRunning` is set whenever both queries
+ // have at least one stage running simultaneously -- which only happens if the two pipelined
+ // groups are genuinely co-scheduled rather than one running after the other.
+ val idA = new AtomicReference[String](null)
+ val idB = new AtomicReference[String](null)
+ val stagesA = ConcurrentHashMap.newKeySet[Int]()
+ val stagesB = ConcurrentHashMap.newKeySet[Int]()
+ val runningA = ConcurrentHashMap.newKeySet[Int]()
+ val runningB = ConcurrentHashMap.newKeySet[Int]()
+ val sawBothRunning = new AtomicBoolean(false)
+ val listener = new SparkListener {
+ override def onJobStart(e: SparkListenerJobStart): Unit = {
+ val qid = Option(e.properties).map(_.getProperty(StreamExecution.QUERY_ID_KEY)).orNull
+ if (qid != null && qid == idA.get()) e.stageIds.foreach(stagesA.add(_))
+ else if (qid != null && qid == idB.get()) e.stageIds.foreach(stagesB.add(_))
+ }
+ override def onStageSubmitted(e: SparkListenerStageSubmitted): Unit = {
+ val sid = e.stageInfo.stageId
+ if (stagesA.contains(sid)) runningA.add(sid)
+ else if (stagesB.contains(sid)) runningB.add(sid)
+ if (!runningA.isEmpty && !runningB.isEmpty) sawBothRunning.set(true)
+ }
+ override def onStageCompleted(e: SparkListenerStageCompleted): Unit = {
+ val sid = e.stageInfo.stageId
+ runningA.remove(sid)
+ runningB.remove(sid)
+ }
+ }
+ spark.sparkContext.addSparkListener(listener)
+
+ // ForeachWriter is one of the sinks RTM allows (see RealTimeModeAllowlist.allowedSinks);
+ // ForeachBatch is not, so it cannot be used to drive a second RTM query here.
+ val handleB = queryB.writeStream
+ .foreach(new ForeachWriter[Row] {
+ override def open(partitionId: Long, epochId: Long): Boolean = true
+ override def process(value: Row): Unit = ()
+ override def close(errorOrNull: Throwable): Unit = ()
+ })
+ .queryName("coexistence_rtm_b")
+ .outputMode(OutputMode.Update)
+ .trigger(defaultTrigger)
+ .start()
+
+ try {
+ idB.set(handleB.id.toString)
+ eventually(timeout(60.seconds)) {
+ assert(handleB.isActive, "second RTM query failed to start")
+ }
+ val execB = handleB.asInstanceOf[StreamingQueryWrapper].streamingQuery
+
+ testStream(queryA, OutputMode.Update, Map.empty, new ContinuousMemorySink())(
+ AddData(inputA, ("x", 1), ("y", 1), ("x", 2)),
+ StartStream(),
+ // StartStream only launches the query; its id is not known until it is running. Record it
+ // here, then drive a SECOND batch below so the listener observes a batch of query A whose
+ // jobs all start after idA is set (the first batch's jobs may fire before this and be
+ // dropped from stagesA). This mirrors the mitigation in the sibling co-scheduling test.
+ Execute(q => idA.set(q.id.toString)),
+ CheckAnswerWithTimeout(60000, "x", "y"),
+ AddData(inputA, ("z", 1), ("w", 1)),
+ CheckAnswerWithTimeout(60000, "x", "y", "z", "w"),
+ Execute { q =>
+ assert(handleB.isActive, "the second RTM query must still be running")
+ assert(handleB.exception.isEmpty,
+ s"second RTM query failed: ${handleB.exception.map(_.getMessage).getOrElse("")}")
+ // Query A's pipelined group is running now. Feed query B so it keeps scheduling its own
+ // group (an idle RTM query can run empty batches), and wait until the listener has seen
+ // a stage of A and a stage of B running at the same time.
+ eventually(timeout(60.seconds)) {
+ inputB.addData(("p", 1), ("q", 1))
+ assert(sawBothRunning.get(),
+ "expected a stage of query A and a stage of query B to run concurrently")
+ }
+ // Both groups are pipelined shuffles (not materialized).
+ assertAllExchangesPipelined(q)
+ assertAllExchangesPipelined(execB)
+ },
+ StopStream
+ )
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ handleB.stop()
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org