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

[stankiewicz]

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]

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.

Footnotes

1. 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. ↩

[stankiewicz]

r: @scwhittle

[github-actions]

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]

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

[scwhittle]

Also adding @tianz101 @jiangzzhu for change streams

[scwhittle]

This has been working well in Radek's tests and looks good to me now. However I would like to get some approval from change stream experts before merging.