[darshan-sj]

Currently, SpannerIO.Write (WriteToSpannerFn.getTransactionOptions()) only forwards Options.priority(...) and Options.maxCommitDelay(...) when calling DatabaseClient.writeAtLeastOnceWithOptions(batch, getTransactionOptions()). It does not expose a way for callers to set:

1. Options.excludeTxnFromChangeStreams() — to exclude write transactions from being recorded in Cloud Spanner change streams configured with the DDL option allow_txn_exclusion = true (e.g., during bulk backfills or forward database migrations so reverse replication pipelines do not re-process migrated writes).
2. Options.tag(String) — to attach a transaction tag (RequestOptions.transaction_tag) to write transactions for Spanner transaction statistics (SPANNER_SYS.TXN_STATS*) and downstream change stream filtering.

Changes

• SpannerConfig:

  • Added getExcludeTxnFromChangeStreams() and withExcludeTxnFromChangeStreams(boolean) / withExcludeTxnFromChangeStreams(ValueProvider<Boolean>).
  • Added getTransactionTag() and withTransactionTag(String) / withTransactionTag(ValueProvider<String>).
  • SpannerIO.Write & WriteToSpannerFn:
  • Added withExcludeTxnFromChangeStreams(...) and withTransactionTag(...) builder methods on SpannerIO.Write.
    • Updated WriteToSpannerFn.getTransactionOptions() to include Options.excludeTxnFromChangeStreams() when excludeTxnFromChangeStreams is true, and Options.tag(...) when transactionTag is non-empty.

• Tests:

  • Extended OptionsImposter with withExcludeTxnFromChangeStreams(Options) and tag(Options).

  • Added runBatchQueryTestWithExcludeTxnFromChangeStreams and runBatchQueryTestWithTransactionTag in SpannerIOWriteTest.

  • End-to-End Dataflow Pipeline Verification:

    • Built and consumed the updated beam-sdks-java-io-google-cloud-platform artifact in the Sourcedb_to_Spanner_Flex Dataflow template and migrated a MySQL table into a Cloud Spanner database configured with CREATE CHANGE STREAM change_stream FOR ALL OPTIONS (allow_txn_exclusion = true).
    • Verified in SPANNER_SYS.TXN_STATS_TOP_MINUTE that all write transactions carried the configured transaction tag (txBy=bulk-<jobId>).
    • Verified via READ_change_stream across all change stream partitions that 0 data_change_records were emitted during the migration.

Checklist

• Spotless formatting applied (./gradlew :sdks:java:io:google-cloud-platform:spotlessApply)
• Unit tests added and passing (./gradlew :sdks:java:io:google-cloud-platform:test --tests org.apache.beam.sdk.io.gcp.spanner.SpannerIOWriteTest)

[github-actions]

Assigning reviewers:

R: @ahmedabu98 for label java.
R: @nielm for label spanner.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[bharadwaj-aditya]

LGTM