[SpannerIO] Add low-latency configuration in Spanner Change Streams by stankiewicz · Pull Request #37580 · apache/beam

stankiewicz · 2026-02-12T14:12:20Z

Background / Goal: The default polling intervals and heartbeat configurations for Spanner change streams (2-minute time increments and 2-second heartbeats) are too slow for pipelines aiming for sub-second latencies. This PR introduces a low-latency mode and improves robustness for continuous change streams reading.

Changes Made:

Added Low-Latency Config Macros: Replaced implicit low-latency flags with explicit cdcTimeIncrement and heartbeatMillis states in SpannerIO.ReadChangeStream.
withLowLatency()
Configuration: Now acts as a macro that configures the change stream pipeline with a 1-second CDC time increment and a 100ms heartbeat interval (defaults are 2 minutes and 2000ms, respectively).
Renamed Variables for Clarity: Renamed legacy configurations like endTimestampDuration to cdcTimeIncrement throughout the action classes and DoFns (QueryChangeStreamAction, ActionFactory, ReadChangeStreamPartitionDoFn).

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

gemini-code-assist · 2026-02-12T14:12:46Z

Summary of Changes

Hello @stankiewicz, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request significantly enhances the Spanner Change Streams connector by introducing explicit low-latency configuration options and improving its resilience against pipeline stalls. The changes allow users to fine-tune polling and heartbeat intervals for sub-second latency requirements and ensure continuous processing even when only heartbeat records are present, leading to a more robust and performant data streaming solution.

Highlights

Low-Latency Configuration: Introduced a new withLowLatency() macro in SpannerIO.ReadChangeStream to easily configure the change stream pipeline for low-latency processing, setting CDC time increment to 1 second and heartbeat interval to 100ms.
Explicit Configuration Parameters: Replaced implicit low-latency flags with explicit cdcTimeIncrement and heartbeatMillis states within SpannerIO.ReadChangeStream and related DoFns, allowing for more granular control over polling and heartbeat intervals.
Improved Pipeline Robustness: Modified HeartbeatRecordAction to return ProcessContinuation.resume() instead of Optional.empty() upon successful heartbeat processing, preventing pipeline stalls when only heartbeat records are emitted during a checkpoint interval.
Clarity and Renaming: Renamed legacy configurations like endTimestampDuration to cdcTimeIncrement across action classes and DoFns for improved clarity and consistency.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Changelog

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
- Added default cdcTimeIncrement and heartbeatMillis to ReadChangeStream builder.
- Introduced abstract getters and setters for cdcTimeIncrement and heartbeatMillis in ReadChangeStream.
- Added a withLowLatency() method to configure low-latency settings.
- Updated InitializeDoFn and ReadChangeStreamPartitionDoFn instantiations to pass new configuration parameters.
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ActionFactory.java
- Modified queryChangeStreamAction method signature and its call to include cdcTimeIncrement.
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/HeartbeatRecordAction.java
- Changed the return value of the run method from Optional.empty() to Optional.of(ProcessContinuation.resume()).
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java
- Added cdcTimeIncrement as a new field and updated the constructor to accept it.
- Modified getNextReadChangeStreamEndTimestamp() to use the configurable cdcTimeIncrement.
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/InitializeDoFn.java
- Replaced the static DEFAULT_HEARTBEAT_MILLIS with an instance field heartbeatMillis.
- Updated the constructor to accept heartbeatMillis and used it when creating PartitionMetadata.
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java
- Added cdcTimeIncrement as a new field and updated the constructor to accept it.
- Modified the setup() method to pass cdcTimeIncrement to the ActionFactory.
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/HeartbeatRecordActionTest.java
- Updated the test assertion to reflect the new ProcessContinuation.resume() return value.
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamActionTest.java
- Updated the QueryChangeStreamAction instantiation in setUp() to pass the cdcTimeIncrement.
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/InitializeDoFnTest.java
- Updated the InitializeDoFn instantiation in setUp() to pass the heartbeatMillis.
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java
- Updated the ReadChangeStreamPartitionDoFn instantiation in setUp() to pass the cdcTimeIncrement.
- Modified the mock setup for actionFactory.queryChangeStreamAction to include cdcTimeIncrement.

Activity

No human activity (comments, reviews) has been recorded for this pull request yet.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

…cdcTimeIncrement` and `heartbeatMillis`; add a low-latency option; and update heartbeat record action to explicitly resume processing.

stankiewicz · 2026-02-12T15:17:36Z

r: @scwhittle

github-actions · 2026-02-12T15:18:54Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

stankiewicz · 2026-02-12T15:59:03Z

r: @chenxuesdu maybe you could take a look at from perspective of change streams

scwhittle · 2026-02-13T10:27:57Z

Also adding @tianz101 @jiangzzhu for change streams

scwhittle · 2026-02-13T10:00:19Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

-  // Return (now + 2 mins) as the end timestamp for reading change streams. This is only used if
-  // users want to run the connector forever. If the end timestamp is reached, we will resume
+  // Return (now + config duration) as the end timestamp for reading change
+  // streams. This is only


nit: wrap better

scwhittle · 2026-02-13T10:10:56Z

...in/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

            throw new IllegalArgumentException("Unknown record type " + record.getClass());
          }

          if (maybeContinuation.isPresent()) {


I think that all previous continuations have been stops. I think that there is an edge case when resuming here we have to confirm is not a potential issue. The timestamp of where we are going to resume processing is updated when we process the heartbeat. However it is possible there is multiple records with the same timestamp. In general the restriction tracker tryClaim makes sure that all records that had the same timestamp are processed to handle this. However with this early return I think if it is possible that a heartbeat has the same timestamp as a record we could exit this processing before we process the record and on resuming we would start past that record.

It seems unlikely that would be the case but I'd rather get some explicit confirmation or handle that.

Perhaps an easy way to handle it would be to notify the interruptor that we want to interrupt when we get a heartbeat. And then it already internally handles ensuring all the timestamps of the same value are processed. We could change the actions to return just a boolean indicating whether or not to stop since the continuation itself isn't well supported.

scwhittle · 2026-02-13T10:14:21Z

...ava/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java

+       * Heartbeat interval for all change stream queries.
+       *
+       * <p>Be careful when changing this interval, as it needs to be less than the checkpointing
+       * interval in Dataflow. Otherwise, if there are no records within checkpoint intervals, the


I previously changed the sdk to ignore runner-initiated splits until tryClaim was made. This ensures some progress per scheduling.

...ava/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java

…dd a low-latency option;

github-actions bot added java io gcp spanner labels Feb 12, 2026

feat: Make Spanner change stream query parameters configurable with `…

b714b84

…cdcTimeIncrement` and `heartbeatMillis`; add a low-latency option; and update heartbeat record action to explicitly resume processing.

stankiewicz force-pushed the spanner_io_configuration branch from de7c40b to fc4cf57 Compare February 12, 2026 14:27

stankiewicz changed the title ~~[SpannerIO] Add low-latency configuration and resume on heartbeats in Spanner Change Streams~~ [SpannerIO] Add low-latency configuration in Spanner Change Streams Feb 13, 2026

scwhittle requested changes Feb 13, 2026

View reviewed changes

Make Spanner change stream query parameters configurable with and ; a…

0d9b313

…dd a low-latency option;

stankiewicz force-pushed the spanner_io_configuration branch from 9f61bbd to 0d9b313 Compare February 13, 2026 16:53

stankiewicz requested a review from scwhittle February 13, 2026 17:55

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SpannerIO] Add low-latency configuration in Spanner Change Streams#37580

[SpannerIO] Add low-latency configuration in Spanner Change Streams#37580
stankiewicz wants to merge 2 commits intoapache:masterfrom
stankiewicz:spanner_io_configuration

stankiewicz commented Feb 12, 2026 •

edited

Loading

Uh oh!

gemini-code-assist bot commented Feb 12, 2026

Uh oh!

stankiewicz commented Feb 12, 2026

Uh oh!

github-actions bot commented Feb 12, 2026

Uh oh!

stankiewicz commented Feb 12, 2026

Uh oh!

scwhittle commented Feb 13, 2026

Uh oh!

scwhittle Feb 13, 2026

Uh oh!

stankiewicz Feb 13, 2026

Uh oh!

scwhittle Feb 13, 2026

Uh oh!

scwhittle Feb 13, 2026

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

stankiewicz commented Feb 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

GitHub Actions Tests Status (on master branch)

Uh oh!

gemini-code-assist bot commented Feb 12, 2026

Summary of Changes

Highlights

Footnotes

Uh oh!

stankiewicz commented Feb 12, 2026

Uh oh!

github-actions bot commented Feb 12, 2026

Uh oh!

stankiewicz commented Feb 12, 2026

Uh oh!

scwhittle commented Feb 13, 2026

Uh oh!

scwhittle Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

stankiewicz Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

scwhittle Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

scwhittle Feb 13, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

stankiewicz commented Feb 12, 2026 •

edited

Loading