[SPARK-58386][SS] Restore backward compatibility of SupportsRealTimeRead.nextWithTimeout by eason-yuchen-liu · Pull Request #57580 · apache/spark

eason-yuchen-liu · 2026-07-27T20:00:39Z

What changes were proposed in this pull request?

Two changes for the public Real-Time Mode (RTM) connector SPI:

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.
Add StreamingRealTimeModeSourceSuite, 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). The source's partition reader intentionally overrides only the legacy single-argument nextWithTimeout(Long), so it is exercised purely through the two-argument default delegation, and the suite runs it end-to-end through a real RTM streaming query.

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.

The new guard suite protects against future regressions of the same kind: because the frozen source uses only public APIs and overrides only the single-arg method, a future source-incompatible change to SupportsRealTimeMode / SupportsRealTimeRead will make the suite fail to compile, and the end-to-end run verifies an external-style source is still driven correctly at runtime.

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?

New test suite StreamingRealTimeModeSourceSuite, which drives a public-API-only RTM source end-to-end through the two-argument default delegation. 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)

uros-b

Thank you @eason-yuchen-liu! Can you just please add automated tests / regression guards to covers the new delegation path?

eason-yuchen-liu · 2026-07-30T21:27:26Z

Thanks. Added a test.

HeartSaVioR

The code change looks OK. We now lost the ability of compile time check which is a major regression, but I understand compatibility breakage is maybe more severe.

My comments are mostly for documentation, especially the clear guidance for 3rd party. We now concern about backward compatibility, implying 3rd party - while we are here, let's avoid confusion to 3rd party and make this be clear to them.

HeartSaVioR · 2026-08-03T02:01:39Z

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.)

HeartSaVioR · 2026-08-03T02:06:27Z

+ * 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 {


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.

HeartSaVioR · 2026-08-03T02:10:08Z

+class StreamingRealTimeModeSourceSuite extends StreamRealTimeModeManualClockSuiteBase {
+  import testImplicits._
+
+  test("RTM source built from only public connector APIs reads end-to-end") {


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.

HeartSaVioR · 2026-08-03T02:11:16Z

+
+/**
+ * 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`


nit: it doesn't need sounds more natural

HeartSaVioR · 2026-08-03T02:12:42Z

+    }
+    // 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.


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.

HyukjinKwon

0 blocking, 1 non-blocking, 0 nits.
Correct, well-tested compatibility restoration; one non-blocking doc suggestion for third-party authors.

Suggestions (1)

sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsRealTimeRead.java:92: Add class-level guidance on which nextWithTimeout overload third parties should implement -- see inline

Verification

Confirmed there is no infinite recursion (2-arg delegates to 1-arg; 1-arg default throws rather than calling back) and that a source overriding neither fails loud with a clear message. The new CompatRealTimePartitionReader overrides only the single-arg overload and is exercised through the engine's two-arg call path, validating the delegation. The loss of compile-time enforcement (both methods now default) is a tradeoff the maintainer explicitly accepted for source compatibility.

HyukjinKwon · 2026-08-04T07:58:23Z

+     * @return {@link RecordStatus} describing whether a record is available and its arrival time
+     * @throws IOException
+     */
+    default RecordStatus nextWithTimeout(Long timeoutMs) throws IOException {


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.)

eason-yuchen-liu · 2026-08-04T19:56:36Z

Thanks for the review. Addressed the comments.

HeartSaVioR

+1 pending CI

HeartSaVioR · 2026-08-05T07:23:33Z

@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.

…ead.nextWithTimeout ### 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 4.1-era sources do) is driven through the delegation. ### 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) Co-authored-by: Isaac

…RTM source SPI ### What changes were proposed in this pull request? Add `StreamingRealTimeModeSourceSuite`, which defines a self-contained Real-Time Mode (RTM) streaming source written exclusively against the public `@Evolving` connector APIs (`TableProvider`, `Table`/`SupportsRead`, `ScanBuilder`/`Scan`, `MicroBatchStream` + `SupportsRealTimeMode`, `PartitionReaderFactory`, `SupportsRealTimeRead`, and custom `Offset`/`PartitionOffset`). The source's partition reader intentionally overrides only the legacy single-argument `nextWithTimeout(Long)` -- the Spark 4.1 contract -- so it is exercised purely through the interface's two-argument default delegation. The test runs the frozen source end-to-end through a real RTM streaming query (via the manual-clock test harness) and asserts the rows are read. ### Why are the changes needed? SPARK-55699 broke source-level backward compatibility of `SupportsRealTimeRead` by replacing an abstract method rather than adding an overload (restored in SPARK-58386). This suite is a guard against future regressions of the same kind: because the frozen source uses only public APIs and overrides only the single-arg method, a future source-incompatible change to `SupportsRealTimeMode` / `SupportsRealTimeRead` will make this file fail to compile, and the end-to-end run verifies an external-style source is still driven correctly at runtime. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New test suite `StreamingRealTimeModeSourceSuite`. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Co-authored-by: Isaac

…oaden compat test ### What changes were proposed in this pull request? Follow-up to review feedback on the RTM connector backward-compatibility change: - `SupportsRealTimeRead`: add interface-level guidance on which `nextWithTimeout` overload third parties should implement, and reword the per-method Javadocs to frame `nextWithTimeout(Long timeoutMs)` as the recommended method and `nextWithTimeout(Long startTimeMs, Long timeoutMs)` as the engine-invoked overload meant for engine-internal sources (the reference clock is not part of the third-party contract). Dropped the "legacy" framing. Docs only; no behavior change. - Rename `StreamingRealTimeModeSourceSuite` to `StreamingRealTimeModeSourceCompatSuite`, reserving the generic name for future general RTM-source tests. - Broaden the guard: extract shared reader logic into a base class and add a second reader that overrides the two-arg `nextWithTimeout` directly, alongside the existing single-arg reader. The source dispatches on a `twoArg` option, and the suite runs both variants end-to-end so the guard tracks either entry point. ### Why are the changes needed? Reviewers asked for clearer third-party guidance on the two overloads and for test coverage of a source that implements the two-arg overload directly, not only the single-arg one reached via default delegation. ### 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` runs a public-API-only RTM source end-to-end for both `nextWithTimeout` entry points (single-arg via default delegation, and two-arg directly). ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Co-authored-by: Isaac

HeartSaVioR · 2026-08-06T05:39:45Z

Thanks! Merging to master/4.x/4.3.

…ead.nextWithTimeout ### 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. - `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. 2. Add `StreamingRealTimeModeSourceSuite`, 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`). The source's partition reader intentionally overrides only the legacy single-argument `nextWithTimeout(Long)`, so it is exercised purely through the two-argument default delegation, and the suite runs it end-to-end through a real RTM streaming query. ### 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. The new guard suite protects against future regressions of the same kind: because the frozen source uses only public APIs and overrides only the single-arg method, a future source-incompatible change to `SupportsRealTimeMode` / `SupportsRealTimeRead` will make the suite fail to compile, and the end-to-end run verifies an external-style source is still driven correctly at runtime. ### 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? New test suite `StreamingRealTimeModeSourceSuite`, which drives a public-API-only RTM source end-to-end through the two-argument default delegation. 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) Closes #57580 from eason-yuchen-liu/SC-234671-restore-nextWithTimeout-compat. Authored-by: Yuchen Liu <170372783+eason-yuchen-liu@users.noreply.github.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com> (cherry picked from commit 041ed93) Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

HeartSaVioR · 2026-08-06T05:42:01Z

Merge Summary:

merged into master 041ed93
merged into branch-4.x 9a4f996
merged into branch-4.3 6e0af39

eason-yuchen-liu · 2026-08-06T17:25:45Z

@HeartSaVioR Thanks! Should we also backport it to 4.2 to fix the regression on that version?

HeartSaVioR · 2026-08-07T12:53:33Z

@eason-yuchen-liu Makes sense. Would you mind helping to create a PR for branch-4.2? Thanks in advance!

uros-b reviewed Jul 28, 2026

View reviewed changes

eason-yuchen-liu requested a review from uros-b July 31, 2026 00:11

HeartSaVioR reviewed Aug 3, 2026

View reviewed changes

HyukjinKwon reviewed Aug 4, 2026

View reviewed changes

eason-yuchen-liu requested review from HeartSaVioR and HyukjinKwon August 4, 2026 19:56

HeartSaVioR approved these changes Aug 4, 2026

View reviewed changes

eason-yuchen-liu added 3 commits August 5, 2026 10:08

eason-yuchen-liu force-pushed the SC-234671-restore-nextWithTimeout-compat branch from 0bd7887 to 8b559a6 Compare August 5, 2026 17:21

eason-yuchen-liu requested a review from HeartSaVioR August 5, 2026 17:21

HeartSaVioR closed this in 041ed93 Aug 6, 2026

Uh oh!

Conversation

eason-yuchen-liu commented Jul 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

uros-b left a comment

Choose a reason for hiding this comment

Uh oh!

eason-yuchen-liu commented Jul 30, 2026

Uh oh!

HeartSaVioR left a comment

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Aug 3, 2026

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon left a comment

Choose a reason for hiding this comment

Suggestions (1)

Verification

Uh oh!

HyukjinKwon Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

eason-yuchen-liu commented Aug 4, 2026

Uh oh!

HeartSaVioR left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Aug 5, 2026

Uh oh!

HeartSaVioR commented Aug 6, 2026

Uh oh!

HeartSaVioR commented Aug 6, 2026

Uh oh!

eason-yuchen-liu commented Aug 6, 2026

Uh oh!

HeartSaVioR commented Aug 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

eason-yuchen-liu commented Jul 27, 2026 •

edited

Loading

HeartSaVioR left a comment •

edited

Loading