This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new c0f276233497 [SPARK-58386][SS] Restore backward compatibility of
SupportsRealTimeRead.nextWithTimeout
c0f276233497 is described below
commit c0f27623349747dc414e4560d0dc9e0cdb11229a
Author: Yuchen Liu <[email protected]>
AuthorDate: Sat Aug 8 13:34:56 2026 +0900
[SPARK-58386][SS] Restore backward compatibility of
SupportsRealTimeRead.nextWithTimeout
Backport of #57580 to branch-4.2. Cherry-picked cleanly from the merged
master commit 041ed937785 (no conflicts).
### What changes were proposed in this pull request?
Two changes for the public Real-Time Mode (RTM) connector SPI:
1. Restore the single-argument `nextWithTimeout(Long timeoutMs)` method on
the `Evolving` `SupportsRealTimeRead` interface, which SPARK-55699 replaced
with `nextWithTimeout(Long startTimeMs, Long timeoutMs)`. Both overloads are
now `default` methods:
- `nextWithTimeout(Long)` throws `UnsupportedOperationException` by
default and is the method a third-party source is expected to implement.
- `nextWithTimeout(Long, Long)` is the overload the engine invokes; by
default it delegates to `nextWithTimeout(Long)`, ignoring `startTimeMs`. It is
meant for engine-internal sources that need the engine's reference start time.
A source overrides whichever one it needs. The engine always invokes the
two-argument overload, so a source that overrides only the single-argument
variant (as external Spark 4.1-era sources do) is driven through the
delegation. Internal implementors (`LowLatencyMemoryStream`, Kafka) already
override the two-argument variant and are unchanged.
2. Add `StreamingRealTimeModeSourceCompatSuite`, a backward-compatibility
guard. It defines a self-contained RTM streaming source written exclusively
against the public `Evolving` connector APIs (`TableProvider`,
`Table`/`SupportsRead`, `ScanBuilder`/`Scan`, `MicroBatchStream` +
`SupportsRealTimeMode`, `PartitionReaderFactory`, `SupportsRealTimeRead`, and
custom `Offset`/`PartitionOffset`) and runs it end-to-end for both
`nextWithTimeout` entry points.
### Why are the changes needed?
SPARK-55699 changed the sole abstract method on the public interface rather
than adding an overload, which is a source- and binary-incompatible break. An
external source (a custom MQTT Real-Time Mode connector) that implemented
`nextWithTimeout(Long)` no longer compiles or links after upgrading from Spark
4.1 to 4.2. The `startTimeMs` parameter is only needed for the engine-internal
manual-clock path, so it should not be forced on external implementors.
### Does this PR introduce _any_ user-facing change?
Yes. It restores the Spark 4.1 `nextWithTimeout(Long)` method on the
`SupportsRealTimeRead` interface, so external sources written against Spark 4.1
compile and link again.
### How was this patch tested?
`StreamingRealTimeModeSourceCompatSuite`, which drives a public-API-only
RTM source end-to-end for both `nextWithTimeout` entry points. Verified it
compiles and passes on branch-4.2.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57860 from eason-yuchen-liu/SC-234671-backport-4.2.
Authored-by: Yuchen Liu
<[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../read/streaming/SupportsRealTimeRead.java | 52 +++-
.../StreamingRealTimeModeSourceCompatSuite.scala | 271 +++++++++++++++++++++
2 files changed, 318 insertions(+), 5 deletions(-)
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsRealTimeRead.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsRealTimeRead.java
index 5542781f333a..50150fc11fee 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsRealTimeRead.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsRealTimeRead.java
@@ -25,7 +25,22 @@ import org.apache.spark.sql.connector.read.PartitionReader;
/**
* A variation on {@link PartitionReader} for use with low latency streaming
processing.
- *
+ * <p>
+ * <b>Which method to implement.</b> A source reader must provide a {@code
nextWithTimeout} that
+ * proceeds to the next record, blocking until one is available or a timeout
elapses. There are two
+ * overloads, and a source should override exactly one:
+ * <ul>
+ * <li>{@link #nextWithTimeout(Long)} -- implement this in the common case.
It receives only the
+ * timeout and is all a third-party source needs.</li>
+ * <li>{@link #nextWithTimeout(Long, Long)} -- the overload the execution
engine actually invokes.
+ * Its default implementation ignores the extra parameter and delegates
to
+ * {@link #nextWithTimeout(Long)}. Override it only for engine-internal
sources that must
+ * observe the engine's reference start time (used together with the
engine-internal low
+ * latency clock); that clock is not part of the third-party contract,
so most sources should
+ * not override this overload.</li>
+ * </ul>
+ * Because both overloads are {@code default} methods, overriding neither is
not caught at compile
+ * time -- it fails at runtime when {@link #nextWithTimeout(Long)} throws.
*/
@Evolving
public interface SupportsRealTimeRead<T> extends PartitionReader<T> {
@@ -78,12 +93,39 @@ public interface SupportsRealTimeRead<T> extends
PartitionReader<T> {
* Alternative function to be called than next(), that proceed to the next
record. The different
* from next() is that, if there is no more records, the call needs to
keep waiting until
* the timeout.
- * @param startTimeMs the base time (milliseconds) the was used to
calculate the timeout.
- * Sources should use it as the reference time to start
waiting for the next
- * record instead of getting the latest time from
LowLatencyClock.
+ * <p>
+ * This is the recommended method to implement for a source. It is enough
for any source that
+ * does not need the engine's reference start time (see {@link
#nextWithTimeout(Long, Long)}).
+ * The engine always invokes {@link #nextWithTimeout(Long, Long)}, whose
default implementation
+ * delegates here, so overriding only this method is sufficient. If a
source overrides neither
+ * overload, this default implementation throws to surface the mistake.
* @param timeoutMs if no result is available after this timeout
(milliseconds), return
* @return {@link RecordStatus} describing whether a record is available
and its arrival time
* @throws IOException
*/
- RecordStatus nextWithTimeout(Long startTimeMs, Long timeoutMs) throws
IOException;
+ default RecordStatus nextWithTimeout(Long timeoutMs) throws IOException {
+ throw new UnsupportedOperationException(
+ "A SupportsRealTimeRead implementation must override either " +
+ "nextWithTimeout(Long) or nextWithTimeout(Long, Long).");
+ }
+
+ /**
+ * The overload of {@link #nextWithTimeout(Long)} that the execution
engine actually invokes. In
+ * addition to the timeout it receives {@code startTimeMs}, the reference
time the engine used
+ * to compute that timeout, so the engine and the source agree on when the
wait started (this
+ * matters when the engine runs against its internal low latency clock,
e.g. a manual clock in
+ * tests). The default implementation ignores {@code startTimeMs} and
delegates to
+ * {@link #nextWithTimeout(Long)}.
+ * <p>
+ * This is intended for engine-internal sources. The reference clock is
not part of the
+ * third-party contract, so third-party sources should implement {@link
#nextWithTimeout(Long)}
+ * and leave this overload to its default.
+ * @param startTimeMs the base time (milliseconds) that was used to
calculate the timeout
+ * @param timeoutMs if no result is available after this timeout
(milliseconds), return
+ * @return {@link RecordStatus} describing whether a record is available
and its arrival time
+ * @throws IOException
+ */
+ default RecordStatus nextWithTimeout(Long startTimeMs, Long timeoutMs)
throws IOException {
+ return nextWithTimeout(timeoutMs);
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingRealTimeModeSourceCompatSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingRealTimeModeSourceCompatSuite.scala
new file mode 100644
index 000000000000..6b20ddf08dac
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingRealTimeModeSourceCompatSuite.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.streaming
+
+import java.util
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.{SupportsRead, Table,
TableCapability, TableProvider}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader,
PartitionReaderFactory, Scan, ScanBuilder}
+import org.apache.spark.sql.connector.read.streaming.{MicroBatchStream,
Offset, PartitionOffset, SupportsRealTimeMode, SupportsRealTimeRead}
+import
org.apache.spark.sql.connector.read.streaming.SupportsRealTimeRead.RecordStatus
+import org.apache.spark.sql.execution.streaming.sources.ContinuousMemorySink
+import org.apache.spark.sql.sources.DataSourceRegister
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.unsafe.types.UTF8String
+
+// scalastyle:off
+/**
+ *
============================================================================================
+ * BACKWARD-COMPATIBILITY GUARD -- DO NOT MODIFY THE SOURCE DEFINITIONS BELOW.
+ *
============================================================================================
+ *
+ * The classes in this file define a self-contained Real-Time Mode (RTM)
streaming source that is
+ * written EXCLUSIVELY against the public, `@Evolving` connector APIs an
external connector author
+ * has access to:
+ *
+ * - `org.apache.spark.sql.sources.DataSourceRegister`
+ * - `org.apache.spark.sql.connector.catalog.{TableProvider, Table,
SupportsRead,
+ * TableCapability}`
+ * - `org.apache.spark.sql.connector.read.{ScanBuilder, Scan,
InputPartition, PartitionReader,
+ * PartitionReaderFactory}`
+ * - `org.apache.spark.sql.connector.read.streaming.{MicroBatchStream,
Offset, PartitionOffset,
+ * SupportsRealTimeMode, SupportsRealTimeRead}`
+ *
+ * It deliberately does NOT use any `private[spark]` / internal helper (no
`LowLatencyMemoryStream`,
+ * `LongOffset`, `SimpleTableProvider`, RPC endpoints, or the engine-internal
low latency clock).
+ * Its purpose is to pin the source-level backward compatibility of the RTM
connector SPI: if a
+ * future change to `SupportsRealTimeMode` or `SupportsRealTimeRead` breaks
external implementors
+ * the way SPARK-55699 did (it replaced `nextWithTimeout(Long)` rather than
adding an overload, see
+ * SPARK-58386), this file will FAIL TO COMPILE, catching the incompatibility
at build time.
+ *
+ * `SupportsRealTimeRead` offers two `nextWithTimeout` overloads and a source
overrides exactly one.
+ * Both are covered here so the guard tracks either entry point:
+ * - `CompatOneArgPartitionReader` overrides only `nextWithTimeout(Long)` --
the method a
+ * third-party source is expected to implement. The engine invokes the
two-arg overload, so this
+ * reader is exercised through the interface's default delegation.
+ * - `CompatTwoArgPartitionReader` overrides `nextWithTimeout(Long, Long)`
directly -- the
+ * overload the engine invokes.
+ * Keep both.
+ *
+ * When you add a genuinely new REQUIRED method to one of these interfaces,
prefer a `default`
+ * method so this frozen source keeps compiling. If a required change is truly
unavoidable, updating
+ * this file is a strong signal that external connectors will also break --
treat it accordingly.
+ *
============================================================================================
+ */
+// scalastyle:on
+
+/** Deterministic, never-changing dataset the guard source serves. */
+private object CompatRealTimeData {
+ val schema: StructType =
+ new StructType().add("value", IntegerType).add("name", StringType)
+
+ // A fixed, finite dataset. Frozen on purpose -- do not change.
+ val records: Array[(Int, String)] = Array((1, "a"), (2, "b"), (3, "c"))
+}
+
+/** A public-API `Offset`: the number of records consumed so far. */
+private case class CompatOffset(consumed: Int) extends Offset {
+ override def json(): String = consumed.toString
+}
+
+/** A public-API per-partition `PartitionOffset`. */
+private case class CompatPartitionOffset(partitionId: Int, offset: Int)
extends PartitionOffset
+
+/** A serializable `InputPartition` carrying its slice of the frozen dataset.
*/
+private case class CompatInputPartition(
+ partitionId: Int,
+ startOffset: Int,
+ rows: Array[(Int, String)])
+ extends InputPartition
+
+/**
+ * Shared reader logic for both `nextWithTimeout` entry points. Concrete
subclasses only pick which
+ * overload of [[SupportsRealTimeRead#nextWithTimeout]] to override; both
route to [[pollNext]].
+ */
+private abstract class CompatRealTimePartitionReaderBase(partition:
CompatInputPartition)
+ extends SupportsRealTimeRead[InternalRow] {
+
+ private var pos = 0
+ private var currentRow: InternalRow = _
+
+ private def toRow(i: Int): InternalRow = {
+ val (v, n) = partition.rows(i)
+ InternalRow(v, UTF8String.fromString(n))
+ }
+
+ /** Return the next record, or wait until the timeout elapses and report no
record. */
+ protected final def pollNext(timeoutMs: java.lang.Long): RecordStatus = {
+ if (pos < partition.rows.length) {
+ val (value, _) = partition.rows(pos)
+ currentRow = toRow(pos)
+ pos += 1
+ // Report the record along with a deterministic synthetic arrival time,
exercising the
+ // arrival-time branch of RecordStatus.
+ return RecordStatus.newStatusWithArrivalTimeMs(value.toLong)
+ }
+ // Exhausted this batch's data: keep waiting until the caller's timeout
elapses, then report
+ // no record -- the same wait-until-timeout behavior a real source has.
Measured against the
+ // wall clock, as a third-party source without the engine's reference
clock would do.
+ val startNs = System.nanoTime()
+ var elapsedMs = 0L
+ while (elapsedMs < timeoutMs) {
+ Thread.sleep(10L)
+ elapsedMs = (System.nanoTime() - startNs) / 1000000L
+ }
+ RecordStatus.newStatusWithoutArrivalTime(false)
+ }
+
+ override def getOffset: PartitionOffset =
+ CompatPartitionOffset(partition.partitionId, partition.startOffset + pos)
+
+ override def next(): Boolean = {
+ if (pos < partition.rows.length) {
+ currentRow = toRow(pos)
+ pos += 1
+ true
+ } else {
+ false
+ }
+ }
+
+ override def get(): InternalRow = currentRow
+
+ override def close(): Unit = {}
+}
+
+/**
+ * Reader that overrides ONLY the single-argument `nextWithTimeout(Long)` --
the method a
+ * third-party source is expected to implement. Reached through the two-arg
default delegation.
+ */
+private class CompatOneArgPartitionReader(partition: CompatInputPartition)
+ extends CompatRealTimePartitionReaderBase(partition) {
+ override def nextWithTimeout(timeoutMs: java.lang.Long): RecordStatus =
pollNext(timeoutMs)
+}
+
+/** Reader that overrides the two-argument `nextWithTimeout(Long, Long)` the
engine invokes. */
+private class CompatTwoArgPartitionReader(partition: CompatInputPartition)
+ extends CompatRealTimePartitionReaderBase(partition) {
+ override def nextWithTimeout(
+ startTimeMs: java.lang.Long, timeoutMs: java.lang.Long): RecordStatus =
pollNext(timeoutMs)
+}
+
+/** A public-API `PartitionReaderFactory`, parameterized by which reader
overload to use. */
+private class CompatRealTimeReaderFactory(twoArg: Boolean) extends
PartitionReaderFactory {
+ override def createReader(partition: InputPartition):
PartitionReader[InternalRow] = {
+ val p = partition.asInstanceOf[CompatInputPartition]
+ if (twoArg) new CompatTwoArgPartitionReader(p) else new
CompatOneArgPartitionReader(p)
+ }
+}
+
+/** The RTM stream: a public `MicroBatchStream` that also implements
`SupportsRealTimeMode`. */
+private class CompatRealTimeStream(twoArg: Boolean)
+ extends MicroBatchStream with SupportsRealTimeMode {
+ override def initialOffset(): Offset = CompatOffset(0)
+ override def deserializeOffset(json: String): Offset =
CompatOffset(json.toInt)
+ override def commit(end: Offset): Unit = {}
+ override def stop(): Unit = {}
+
+ override def latestOffset(): Offset =
CompatOffset(CompatRealTimeData.records.length)
+ override def planInputPartitions(start: Offset, end: Offset):
Array[InputPartition] = {
+ val from = start.asInstanceOf[CompatOffset].consumed
+ val to = end.asInstanceOf[CompatOffset].consumed
+ Array(CompatInputPartition(0, from, CompatRealTimeData.records.slice(from,
to)))
+ }
+ override def createReaderFactory(): PartitionReaderFactory =
+ new CompatRealTimeReaderFactory(twoArg)
+
+ override def planInputPartitions(start: Offset): Array[InputPartition] = {
+ val from = start.asInstanceOf[CompatOffset].consumed
+ Array(CompatInputPartition(0, from, CompatRealTimeData.records.drop(from)))
+ }
+ override def mergeOffsets(offsets: Array[PartitionOffset]): Offset = {
+ val maxOffset =
offsets.map(_.asInstanceOf[CompatPartitionOffset].offset).max
+ CompatOffset(maxOffset)
+ }
+}
+
+/** Scan + ScanBuilder wired to the RTM stream, using only public APIs. */
+private class CompatRealTimeScan(twoArg: Boolean) extends ScanBuilder with
Scan {
+ override def build(): Scan = this
+ override def readSchema(): StructType = CompatRealTimeData.schema
+ override def toMicroBatchStream(checkpointLocation: String):
MicroBatchStream =
+ new CompatRealTimeStream(twoArg)
+}
+
+/**
+ * The top-level source, registered as a `TableProvider` +
`DataSourceRegister`. Loaded by fully
+ * qualified class name via `spark.readStream.format(...)`, so it doesn't need
a `META-INF/services`
+ * registration entry. The `twoArg` option selects which `nextWithTimeout`
overload the reader
+ * implements.
+ */
+class CompatRealTimeSourceProvider extends TableProvider with
DataSourceRegister {
+ override def shortName(): String = "compat-realtime-source"
+
+ override def inferSchema(options: CaseInsensitiveStringMap): StructType =
+ CompatRealTimeData.schema
+
+ override def getTable(
+ schema: StructType,
+ partitioning: Array[Transform],
+ properties: util.Map[String, String]): Table = new CompatRealTimeTable
+}
+
+private class CompatRealTimeTable extends Table with SupportsRead {
+ override def name(): String = "compat-realtime-source"
+ override def schema(): StructType = CompatRealTimeData.schema
+ override def capabilities(): util.Set[TableCapability] =
+ util.EnumSet.of(TableCapability.MICRO_BATCH_READ)
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder =
+ new CompatRealTimeScan(options.getBoolean("twoArg", false))
+}
+
+/**
+ * SPARK-58386: a compile-time and runtime backward-compatibility guard for
the public Real-Time
+ * Mode connector SPI (`SupportsRealTimeMode` / `SupportsRealTimeRead`). The
frozen source above
+ * must keep compiling against these interfaces, and these tests run it
end-to-end through a real
+ * RTM streaming query to prove an external-style source is still driven
correctly, for both
+ * `nextWithTimeout` entry points.
+ */
+class StreamingRealTimeModeSourceCompatSuite extends
StreamRealTimeModeManualClockSuiteBase {
+ import testImplicits._
+
+ private def runSourceEndToEnd(twoArg: Boolean): Unit = {
+ val df = spark.readStream
+ .format(classOf[CompatRealTimeSourceProvider].getName)
+ .option("twoArg", twoArg)
+ .load()
+ .selectExpr("concat(cast(value as string), '-', name) as output")
+
+ testStream(df, OutputMode.Update, Map.empty, new ContinuousMemorySink())(
+ StartStream(),
+ CheckAnswerWithTimeout(10000, "1-a", "2-b", "3-c"),
+ StopStream
+ )
+ }
+
+ test("RTM source implementing single-arg nextWithTimeout reads end-to-end") {
+ runSourceEndToEnd(twoArg = false)
+ }
+
+ test("RTM source implementing two-arg nextWithTimeout reads end-to-end") {
+ runSourceEndToEnd(twoArg = true)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
