This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 1ed19f8d230 [SpannerIO] Support excludeTxnFromChangeStreams and
transactionTag in SpannerIO.Write (#40489)
1ed19f8d230 is described below
commit 1ed19f8d23063ebd7d416896b20ba3f41a482da2
Author: darshan-sj <[email protected]>
AuthorDate: Thu Oct 15 18:48:18 2026 +0000
[SpannerIO] Support excludeTxnFromChangeStreams and transactionTag in
SpannerIO.Write (#40489)
* [SpannerIO] Support excludeTxnFromChangeStreams and transactionTag in
SpannerIO.Write
* Added negative test case
* Formatting changes
---
.../beam/sdk/io/gcp/spanner/SpannerConfig.java | 37 +++++++++++
.../apache/beam/sdk/io/gcp/spanner/SpannerIO.java | 45 ++++++++++++-
.../com/google/cloud/spanner/OptionsImposter.java | 8 +++
.../sdk/io/gcp/spanner/SpannerIOWriteTest.java | 76 ++++++++++++++++++++++
4 files changed, 165 insertions(+), 1 deletion(-)
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerConfig.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerConfig.java
index 36b13f237b4..0ca21debb59 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerConfig.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerConfig.java
@@ -117,6 +117,10 @@ public abstract class SpannerConfig implements
Serializable {
public abstract @Nullable ValueProvider<Boolean>
getEnableOpenTelemetryTracing();
+ public abstract @Nullable ValueProvider<Boolean>
getExcludeTxnFromChangeStreams();
+
+ public abstract @Nullable ValueProvider<String> getTransactionTag();
+
abstract Builder toBuilder();
public static SpannerConfig create() {
@@ -210,6 +214,11 @@ public abstract class SpannerConfig implements
Serializable {
abstract Builder setEnableOpenTelemetryTracing(
ValueProvider<Boolean> enableOpenTelemetryTracing);
+ abstract Builder setExcludeTxnFromChangeStreams(
+ ValueProvider<Boolean> excludeTxnFromChangeStreams);
+
+ abstract Builder setTransactionTag(ValueProvider<String> transactionTag);
+
abstract Builder setClientCertPath(ValueProvider<String> clientCertPath);
abstract Builder setClientCertKeyPath(ValueProvider<String>
clientCertKeyPath);
@@ -479,6 +488,34 @@ public abstract class SpannerConfig implements
Serializable {
ValueProvider.StaticValueProvider.of(enableOpenTelemetryTracing));
}
+ /**
+ * Specifies whether transactions should be excluded from being recorded in
change streams with
+ * the DDL option {@code allow_txn_exclusion=true}.
+ */
+ public SpannerConfig withExcludeTxnFromChangeStreams(
+ ValueProvider<Boolean> excludeTxnFromChangeStreams) {
+ return
toBuilder().setExcludeTxnFromChangeStreams(excludeTxnFromChangeStreams).build();
+ }
+
+ /**
+ * Specifies whether transactions should be excluded from being recorded in
change streams with
+ * the DDL option {@code allow_txn_exclusion=true}.
+ */
+ public SpannerConfig withExcludeTxnFromChangeStreams(boolean
excludeTxnFromChangeStreams) {
+ return withExcludeTxnFromChangeStreams(
+ ValueProvider.StaticValueProvider.of(excludeTxnFromChangeStreams));
+ }
+
+ /** Specifies the transaction tag to use for write transactions. */
+ public SpannerConfig withTransactionTag(ValueProvider<String>
transactionTag) {
+ return toBuilder().setTransactionTag(transactionTag).build();
+ }
+
+ /** Specifies the transaction tag to use for write transactions. */
+ public SpannerConfig withTransactionTag(String transactionTag) {
+ return
withTransactionTag(ValueProvider.StaticValueProvider.of(transactionTag));
+ }
+
/**
* Specifies certificate paths to use for mTLS channel.
*
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
index 4c709963fc6..cc573686357 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
@@ -1616,6 +1616,41 @@ public class SpannerIO {
return withSpannerConfig(config.withMaxCommitDelay(millis));
}
+ /**
+ * Specifies whether transactions should be excluded from being recorded
in change streams with
+ * the DDL option {@code allow_txn_exclusion=true}. This does not exclude
the transaction from
+ * being recorded in the change streams with the DDL option {@code
allow_txn_exclusion} being
+ * false or unset.
+ */
+ public Write withExcludeTxnFromChangeStreams(boolean
excludeTxnFromChangeStreams) {
+ SpannerConfig config = getSpannerConfig();
+ return
withSpannerConfig(config.withExcludeTxnFromChangeStreams(excludeTxnFromChangeStreams));
+ }
+
+ /**
+ * Specifies whether transactions should be excluded from being recorded
in change streams with
+ * the DDL option {@code allow_txn_exclusion=true}. This does not exclude
the transaction from
+ * being recorded in the change streams with the DDL option {@code
allow_txn_exclusion} being
+ * false or unset.
+ */
+ public Write withExcludeTxnFromChangeStreams(
+ ValueProvider<Boolean> excludeTxnFromChangeStreams) {
+ SpannerConfig config = getSpannerConfig();
+ return
withSpannerConfig(config.withExcludeTxnFromChangeStreams(excludeTxnFromChangeStreams));
+ }
+
+ /** Specifies the transaction tag to use for write transactions. */
+ public Write withTransactionTag(String transactionTag) {
+ SpannerConfig config = getSpannerConfig();
+ return withSpannerConfig(config.withTransactionTag(transactionTag));
+ }
+
+ /** Specifies the transaction tag to use for write transactions. */
+ public Write withTransactionTag(ValueProvider<String> transactionTag) {
+ SpannerConfig config = getSpannerConfig();
+ return withSpannerConfig(config.withTransactionTag(transactionTag));
+ }
+
/**
* Specifies the maximum cumulative backoff time when retrying after
DEADLINE_EXCEEDED errors.
* Default is 15 mins.
@@ -3063,7 +3098,15 @@ public class SpannerIO {
java.time.Duration.ofMillis(
spannerConfig.getMaxCommitDelay().get().getMillis()))
: null,
- commitStats ? Options.commitStats() : null)
+ commitStats ? Options.commitStats() : null,
+ spannerConfig.getExcludeTxnFromChangeStreams() != null
+ &&
Boolean.TRUE.equals(spannerConfig.getExcludeTxnFromChangeStreams().get())
+ ? Options.excludeTxnFromChangeStreams()
+ : null,
+ spannerConfig.getTransactionTag() != null
+ &&
!Strings.isNullOrEmpty(spannerConfig.getTransactionTag().get())
+ ? Options.tag(spannerConfig.getTransactionTag().get())
+ : null)
.filter(Objects::nonNull)
.toArray(Options.TransactionOption[]::new);
}
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/com/google/cloud/spanner/OptionsImposter.java
b/sdks/java/io/google-cloud-platform/src/test/java/com/google/cloud/spanner/OptionsImposter.java
index 72fc3a0d3d2..9d8b87b4445 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/com/google/cloud/spanner/OptionsImposter.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/com/google/cloud/spanner/OptionsImposter.java
@@ -36,4 +36,12 @@ public class OptionsImposter {
public static CommitResponse
createCommitResponse(com.google.spanner.v1.CommitResponse proto) {
return new CommitResponse(proto);
}
+
+ public static Boolean withExcludeTxnFromChangeStreams(Options options) {
+ return options.withExcludeTxnFromChangeStreams();
+ }
+
+ public static String tag(Options options) {
+ return options.tag();
+ }
}
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOWriteTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOWriteTest.java
index cf8b876d039..ab227a9a239 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOWriteTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOWriteTest.java
@@ -24,6 +24,7 @@ import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.hasSize;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -325,6 +326,81 @@ public class SpannerIOWriteTest implements Serializable {
}));
}
+ @Test
+ public void runBatchQueryTestWithExcludeTxnFromChangeStreams() {
+ SpannerIO.Write write =
+ SpannerIO.write()
+ .withSpannerConfig(SPANNER_CONFIG)
+ .withServiceFactory(serviceFactory)
+ .withExcludeTxnFromChangeStreams(true);
+
assertTrue(write.getSpannerConfig().getExcludeTxnFromChangeStreams().get());
+
+ Mutation mutation = buildUpsertMutation(2L);
+ PCollection<Mutation> mutations = pipeline.apply(Create.of(mutation));
+ mutations.apply(write);
+ pipeline.run();
+
+ verify(serviceFactory.mockDatabaseClient(), times(1))
+ .writeAtLeastOnceWithOptions(
+ mutationsInNoOrder(buildMutationBatch(mutation)),
+ any(ReadQueryUpdateTransactionOption.class),
+ argThat(
+ opts -> {
+ Options options =
OptionsImposter.fromTransactionOptions(opts);
+ return Boolean.TRUE.equals(
+
OptionsImposter.withExcludeTxnFromChangeStreams(options));
+ }));
+ }
+
+ @Test
+ public void runBatchQueryTestWithExcludeTxnFromChangeStreamsFalse() {
+ SpannerIO.Write write =
+ SpannerIO.write()
+ .withSpannerConfig(SPANNER_CONFIG)
+ .withServiceFactory(serviceFactory)
+ .withExcludeTxnFromChangeStreams(false);
+
assertFalse(write.getSpannerConfig().getExcludeTxnFromChangeStreams().get());
+
+ Mutation mutation = buildUpsertMutation(2L);
+ PCollection<Mutation> mutations = pipeline.apply(Create.of(mutation));
+ mutations.apply(write);
+ pipeline.run();
+
+ verify(serviceFactory.mockDatabaseClient(), times(1))
+ .writeAtLeastOnceWithOptions(
+ mutationsInNoOrder(buildMutationBatch(mutation)),
+ argThat(
+ (Options.TransactionOption opt) -> {
+ Options options =
OptionsImposter.fromTransactionOptions(opt);
+ return
OptionsImposter.withExcludeTxnFromChangeStreams(options) == null;
+ }));
+ }
+
+ @Test
+ public void runBatchQueryTestWithTransactionTag() {
+ SpannerIO.Write write =
+ SpannerIO.write()
+ .withSpannerConfig(SPANNER_CONFIG)
+ .withServiceFactory(serviceFactory)
+ .withTransactionTag("txBy=testJobId");
+ assertEquals("txBy=testJobId",
write.getSpannerConfig().getTransactionTag().get());
+
+ Mutation mutation = buildUpsertMutation(2L);
+ PCollection<Mutation> mutations = pipeline.apply(Create.of(mutation));
+ mutations.apply(write);
+ pipeline.run();
+
+ verify(serviceFactory.mockDatabaseClient(), times(1))
+ .writeAtLeastOnceWithOptions(
+ mutationsInNoOrder(buildMutationBatch(mutation)),
+ any(ReadQueryUpdateTransactionOption.class),
+ argThat(
+ opts -> {
+ Options options =
OptionsImposter.fromTransactionOptions(opts);
+ return "txBy=testJobId".equals(OptionsImposter.tag(options));
+ }));
+ }
+
@Test
public void singleMutationPipeline() throws Exception {
Mutation mutation = buildUpsertMutation(2L);
