This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 0bf401810d99 [SPARK-58698][SS][TEST] Add additional tests for RTM
0bf401810d99 is described below
commit 0bf401810d997a7988863c8b2c78f318277da392
Author: Boyang Jerry Peng <[email protected]>
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 <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
