You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "eason-yuchen-liu (via GitHub)" <gi...@apache.org> on 2026/07/27 20:00:42 UTC
[PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
eason-yuchen-liu opened a new pull request, #57580:
URL: https://github.com/apache/spark/pull/57580
### What changes were proposed in this pull request?
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.
- `nextWithTimeout(Long, Long)` delegates to `nextWithTimeout(Long)` by default, ignoring `startTimeMs`.
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.
### 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 manual-clock/test 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?
Verified locally that a Real-Time Mode source implementing only the single-argument `nextWithTimeout(Long)` is driven correctly end-to-end by a streaming query through the two-argument default delegation, and that existing RTM suites (`StreamRealTimeModeSuite`, `StreamRealTimeModeWithManualClockSuite`, `StreamRealTimeModeE2ESuite`) continue to pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on PR #57580:
URL: https://github.com/apache/spark/pull/57580#issuecomment-5183929846
Thanks for the review. Addressed the comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #57580:
URL: https://github.com/apache/spark/pull/57580#issuecomment-5200819477
Thanks! Merging to master/4.x/4.3.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on PR #57580:
URL: https://github.com/apache/spark/pull/57580#issuecomment-5136440794
Thanks. Added a test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #57580:
URL: https://github.com/apache/spark/pull/57580#discussion_r3700926242
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsRealTimeRead.java:
##########
Review Comment:
This is getting complicated for 3rd party data source implementations. Can we revisit the explanation from 3rd party perspective, and provide the high-level guide in classdoc? Is LowLatencyClock exposed to 3rd party? How they would have tested on their data source implementation?
I'd rather give more direct guidance about which one to use.
If 3rd party isn't expected to leverage LowLatencyClock, we should have guided them to use only `RecordStatus nextWithTimeout(Long timeoutMs) throws IOException` and describe `default RecordStatus nextWithTimeout(Long startTimeMs, Long timeoutMs) throws IOException` as internal usage.
If we think LowLatencyClock is a must for 3rd party to test their data source implementation, I'd rather guide `RecordStatus nextWithTimeout(Long timeoutMs) throws IOException` to be a legacy and encourage to implement `default RecordStatus nextWithTimeout(Long startTimeMs, Long timeoutMs) throws IOException`.
(Though the new test suite clarifies that LowLatencyClock isn't meant to be used from 3rd party data source implementations, so this is unlikely true.)
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingRealTimeModeSourceSuite.scala:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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 `LowLatencyClock`). 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.
+ *
+ * In particular, `CompatRealTimePartitionReader` overrides ONLY the single-argument
+ * `nextWithTimeout(Long)` -- exactly how a pre-SPARK-55699 (Spark 4.1) source is written. The
+ * engine only ever calls the two-argument overload, so this reader is exercised purely through the
+ * interface's default delegation. Keep it that way.
+ *
+ * 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
+
+/**
+ * Partition reader implementing ONLY the legacy single-argument
+ * [[SupportsRealTimeRead#nextWithTimeout(Long)]] -- the Spark 4.1 contract. See the file header.
+ */
+private class CompatRealTimePartitionReader(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))
+ }
+
+ override def nextWithTimeout(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, since this legacy overload is not given the engine's reference start time.
+ 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 = {}
+}
+
+/** A public-API `PartitionReaderFactory`. */
+private object CompatRealTimeReaderFactory extends PartitionReaderFactory {
+ override def createReader(partition: InputPartition): PartitionReader[InternalRow] =
+ new CompatRealTimePartitionReader(partition.asInstanceOf[CompatInputPartition])
+}
+
+/** The RTM stream: a public `MicroBatchStream` that also implements `SupportsRealTimeMode`. */
+private class CompatRealTimeStream 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 = CompatRealTimeReaderFactory
+
+ 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 extends ScanBuilder with Scan {
+ override def build(): Scan = this
+ override def readSchema(): StructType = CompatRealTimeData.schema
+ override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream =
+ new CompatRealTimeStream
+}
+
+/**
+ * The top-level source, registered as a `TableProvider` + `DataSourceRegister`. Loaded by fully
+ * qualified class name via `spark.readStream.format(...)`, so it needs no `META-INF/services`
Review Comment:
nit: `it doesn't need` sounds more natural
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingRealTimeModeSourceSuite.scala:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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 `LowLatencyClock`). 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.
+ *
+ * In particular, `CompatRealTimePartitionReader` overrides ONLY the single-argument
+ * `nextWithTimeout(Long)` -- exactly how a pre-SPARK-55699 (Spark 4.1) source is written. The
+ * engine only ever calls the two-argument overload, so this reader is exercised purely through the
+ * interface's default delegation. Keep it that way.
+ *
+ * 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
+
+/**
+ * Partition reader implementing ONLY the legacy single-argument
+ * [[SupportsRealTimeRead#nextWithTimeout(Long)]] -- the Spark 4.1 contract. See the file header.
+ */
+private class CompatRealTimePartitionReader(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))
+ }
+
+ override def nextWithTimeout(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, since this legacy overload is not given the engine's reference start time.
Review Comment:
nit: `this legacy overload` this is totally dependent on which method would be the main entry for 3rd party. If they are expected to use one param, it's not a legacy thing.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingRealTimeModeSourceSuite.scala:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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 `LowLatencyClock`). 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.
+ *
+ * In particular, `CompatRealTimePartitionReader` overrides ONLY the single-argument
+ * `nextWithTimeout(Long)` -- exactly how a pre-SPARK-55699 (Spark 4.1) source is written. The
+ * engine only ever calls the two-argument overload, so this reader is exercised purely through the
+ * interface's default delegation. Keep it that way.
+ *
+ * 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
+
+/**
+ * Partition reader implementing ONLY the legacy single-argument
+ * [[SupportsRealTimeRead#nextWithTimeout(Long)]] -- the Spark 4.1 contract. See the file header.
+ */
+private class CompatRealTimePartitionReader(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))
+ }
+
+ override def nextWithTimeout(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, since this legacy overload is not given the engine's reference start time.
+ 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 = {}
+}
+
+/** A public-API `PartitionReaderFactory`. */
+private object CompatRealTimeReaderFactory extends PartitionReaderFactory {
+ override def createReader(partition: InputPartition): PartitionReader[InternalRow] =
+ new CompatRealTimePartitionReader(partition.asInstanceOf[CompatInputPartition])
+}
+
+/** The RTM stream: a public `MicroBatchStream` that also implements `SupportsRealTimeMode`. */
+private class CompatRealTimeStream 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 = CompatRealTimeReaderFactory
+
+ 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 extends ScanBuilder with Scan {
+ override def build(): Scan = this
+ override def readSchema(): StructType = CompatRealTimeData.schema
+ override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream =
+ new CompatRealTimeStream
+}
+
+/**
+ * The top-level source, registered as a `TableProvider` + `DataSourceRegister`. Loaded by fully
+ * qualified class name via `spark.readStream.format(...)`, so it needs no `META-INF/services`
+ * registration entry.
+ */
+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
+}
+
+/**
+ * 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 this test runs it end-to-end through a real
+ * RTM streaming query to prove an external-style source is still driven correctly.
+ */
+class StreamingRealTimeModeSourceSuite extends StreamRealTimeModeManualClockSuiteBase {
Review Comment:
I'd add `Compat` (or `Compatibility` as full representation) in the test suite name explicitly; we may want to have some tests in RTM source in general in future.
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingRealTimeModeSourceSuite.scala:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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 `LowLatencyClock`). 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.
+ *
+ * In particular, `CompatRealTimePartitionReader` overrides ONLY the single-argument
+ * `nextWithTimeout(Long)` -- exactly how a pre-SPARK-55699 (Spark 4.1) source is written. The
+ * engine only ever calls the two-argument overload, so this reader is exercised purely through the
+ * interface's default delegation. Keep it that way.
+ *
+ * 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
+
+/**
+ * Partition reader implementing ONLY the legacy single-argument
+ * [[SupportsRealTimeRead#nextWithTimeout(Long)]] -- the Spark 4.1 contract. See the file header.
+ */
+private class CompatRealTimePartitionReader(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))
+ }
+
+ override def nextWithTimeout(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, since this legacy overload is not given the engine's reference start time.
+ 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 = {}
+}
+
+/** A public-API `PartitionReaderFactory`. */
+private object CompatRealTimeReaderFactory extends PartitionReaderFactory {
+ override def createReader(partition: InputPartition): PartitionReader[InternalRow] =
+ new CompatRealTimePartitionReader(partition.asInstanceOf[CompatInputPartition])
+}
+
+/** The RTM stream: a public `MicroBatchStream` that also implements `SupportsRealTimeMode`. */
+private class CompatRealTimeStream 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 = CompatRealTimeReaderFactory
+
+ 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 extends ScanBuilder with Scan {
+ override def build(): Scan = this
+ override def readSchema(): StructType = CompatRealTimeData.schema
+ override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream =
+ new CompatRealTimeStream
+}
+
+/**
+ * The top-level source, registered as a `TableProvider` + `DataSourceRegister`. Loaded by fully
+ * qualified class name via `spark.readStream.format(...)`, so it needs no `META-INF/services`
+ * registration entry.
+ */
+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
+}
+
+/**
+ * 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 this test runs it end-to-end through a real
+ * RTM streaming query to prove an external-style source is still driven correctly.
+ */
+class StreamingRealTimeModeSourceSuite extends StreamRealTimeModeManualClockSuiteBase {
+ import testImplicits._
+
+ test("RTM source built from only public connector APIs reads end-to-end") {
Review Comment:
Don't we want to have two params as well? I know internal data sources use two params so it sounds like redundant, but maybe good to be explicit. Maybe we could reuse a lot of code for one param version of classes, via inheritance, or even just source option to dispatch the class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #57580:
URL: https://github.com/apache/spark/pull/57580#discussion_r3710507495
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsRealTimeRead.java:
##########
@@ -78,12 +78,40 @@ public Optional<Long> recArrivalTime() {
* 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>
+ * A source must override exactly one of {@link #nextWithTimeout(Long)} and
+ * {@link #nextWithTimeout(Long, Long)}. This single-argument variant is enough for sources that
+ * do not need the engine's reference start time. The engine always invokes
+ * {@link #nextWithTimeout(Long, Long)}, whose default implementation delegates here, so
+ * overriding only this method is sufficient. The default implementation of this method throws,
+ * since the two-argument variant must have been overridden if this one was not.
+ * @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 timeoutMs) throws IOException {
Review Comment:
A class-level note would help third-party source authors: implement `nextWithTimeout(Long timeoutMs)` for the common case, and only override the two-arg `nextWithTimeout(Long startTimeMs, Long timeoutMs)` if you need the engine's reference start time (`LowLatencyClock` is engine-internal, not part of the third-party contract). The per-method javadoc already says "override exactly one", but since both are now `default` the "must implement one" check only fails at runtime, so the high-level "which one and why" up front is worth it. (Raised by @HeartSaVioR as well.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #57580:
URL: https://github.com/apache/spark/pull/57580#issuecomment-5200834223
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/041ed93778521a6a270fb74661badd3b336ee592
- merged into branch-4.x https://github.com/apache/spark/commit/9a4f99649c2aa340477082f3680e8ac2e01e11fa
- merged into branch-4.3 https://github.com/apache/spark/commit/6e0af391a56e57943ab1f0bc30a21a593cded33e
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on PR #57580:
URL: https://github.com/apache/spark/pull/57580#issuecomment-5207802841
@HeartSaVioR Thanks! Should we also backport it to 4.2 to fix the regression on that version?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on PR #57580:
URL: https://github.com/apache/spark/pull/57580#issuecomment-5221468023
Sure: https://github.com/apache/spark/pull/57860
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #57580:
URL: https://github.com/apache/spark/pull/57580#issuecomment-5188819183
@eason-yuchen-liu
Could you please rebase this with master branch? There is a failure on CI especially "base build image" and it will be fixed with latest master branch IIUC.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #57580: [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout
URL: https://github.com/apache/spark/pull/57580
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #57580:
URL: https://github.com/apache/spark/pull/57580#issuecomment-5217286322
@eason-yuchen-liu Makes sense. Would you mind helping to create a PR for branch-4.2? Thanks in advance!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org