docs(samples): add samples and tests for change streams transaction exclusion

samples/snippets/pom.xml

@@ -34,7 +34,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.37.0</version>
+        <version>26.38.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>

samples/snippets/src/main/java/com/example/spanner/ChangeStreamsTxnExclusionSample.java

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example.spanner;
+
+import com.google.cloud.spanner.CommitResponse;
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.DatabaseId;
+import com.google.cloud.spanner.Mutation;
+import com.google.cloud.spanner.Options;
+import com.google.cloud.spanner.Spanner;
+import com.google.cloud.spanner.SpannerOptions;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.TransactionContext;
+import com.google.cloud.spanner.TransactionManager;
+import java.util.Collections;
+import com.google.spanner.v1.BatchWriteResponse;
+import com.google.api.gax.rpc.ServerStream;
+import com.google.rpc.Code;
+import com.google.cloud.spanner.MutationGroup;
+import com.google.common.collect.ImmutableList;
+
+/** Sample showing how to set exclude transaction from change streams in different write requests. */
+public class ChangeStreamsTxnExclusionSample {
+
+  static void setExcludeTxnFromChangeStreams() {
+    // TODO(developer): Replace these variables before running the sample.
+    final String projectId = "span-cloud-testing";
+    final String instanceId = "weideng-test";
+    final String databaseId = "my-database";
+
+    try (Spanner spanner =
+        SpannerOptions.newBuilder().setProjectId(projectId).build().getService()) {
+      final DatabaseClient databaseClient =
+          spanner.getDatabaseClient(DatabaseId.of(projectId, instanceId, databaseId));
+      rwTxnExcludedFromChangeStreams(databaseClient);
+    }
+  }
+
+  // [START spanner_set_exclude_txn_from_change_streams]
+  static void rwTxnExcludedFromChangeStreams(DatabaseClient client) {
+    // Exclude the transaction from allowed tracking change streams with alloww_txn_exclusion=true.
+    // This exclusion will be applied to all the individual operations inside this transaction.
+    client
+        .readWriteTransaction(Options.excludeTxnFromChangeStreams())
+        .run(
+            transaction -> {
+              transaction.executeUpdate(
+                  Statement.of(
+                      "INSERT Singers (SingerId, FirstName, LastName)\n"
+                          + "VALUES (1341, 'Virginia', 'Watson')"));
+              System.out.println("New singer inserted.");
+
+              transaction.executeUpdate(
+                  Statement.of("UPDATE Singers SET FirstName = 'Hi' WHERE SingerId = 111"));
+              System.out.println("Singer first name updated.");
+
+              return null;
+            });
+  }
+
+  static void writeExcludedFromChangeStreams(DatabaseClient client) {
+    CommitResponse response =
+        client.writeWithOptions(
+            Collections.singletonList(
+                Mutation.newInsertOrUpdateBuilder("Singers")
+                    .set("SingerId")
+                    .to(4520)
+                    .set("FirstName")
+                    .to("Lauren")
+                    .set("LastName")
+                    .to("Lee")
+                    .build()),
+            Options.excludeTxnFromChangeStreams());
+    System.out.println("New singer inserted.");
+  }
+
+  static void writeAtLeastOnceExcludedFromChangeStreams(DatabaseClient client) {
+    CommitResponse response =
+        client.writeAtLeastOnceWithOptions(
+            Collections.singletonList(
+                Mutation.newInsertOrUpdateBuilder("Singers")
+                    .set("SingerId")
+                    .to(45201)
+                    .set("FirstName")
+                    .to("Laura")
+                    .set("LastName")
+                    .to("Johnson")
+                    .build()),
+            Options.excludeTxnFromChangeStreams());
+    System.out.println("New singer inserted.");
+  }
+
+  static void batchWriteAtLeastOnceExcludedFromChangeStreams(DatabaseClient client) {
+    ServerStream<BatchWriteResponse> responses =
+        client.batchWriteAtLeastOnce(
+            ImmutableList.of(MutationGroup.of(
+                Mutation.newInsertOrUpdateBuilder("Singers")
+                    .set("SingerId")
+                    .to(116)
+                    .set("FirstName")
+                    .to("Scarlet")
+                    .set("LastName")
+                    .to("Terry")
+                    .build())),
+            Options.excludeTxnFromChangeStreams());
+    for (BatchWriteResponse response : responses) {
+      if (response.getStatus().getCode() == Code.OK_VALUE) {
+        System.out.printf(
+            "Mutation group have been applied with commit timestamp %s",
+            response.getIndexesList(), response.getCommitTimestamp());
+      } else {
+        System.out.printf(
+            "Mutation group could not be applied with error code %s and "
+                + "error message %s", response.getIndexesList(),
+            Code.forNumber(response.getStatus().getCode()), response.getStatus().getMessage());
+      }
+    }
+  }
+
+  static void pdmlExcludedFromChangeStreams(DatabaseClient client) {
+    client.executePartitionedUpdate(
+        Statement.of("DELETE FROM Singers WHERE TRUE"), Options.excludeTxnFromChangeStreams());
+    System.out.println("Singers deleted.");
+  }
+
+  static void txnManagerExcludedFromChangeStreams(DatabaseClient client) {
+    try (TransactionManager manager = client.transactionManager(Options.excludeTxnFromChangeStreams())) {
+      TransactionContext transaction = manager.begin();
+      transaction.buffer(
+          Mutation.newInsertOrUpdateBuilder("Singers")
+              .set("SingerId")
+              .to(888)
+              .set("FirstName")
+              .to("Johnson")
+              .set("LastName")
+              .to("Doug")
+              .build());
+      manager.commit();
+      System.out.println("New singer inserted.");
+    }
+  }
+
+  // [END spanner_set_exclude_txn_from_change_streams]
+
+}

samples/snippets/src/test/java/com/example/spanner/ChangeStreamsTxnExclusionSampleIT.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example.spanner;
+
+import static com.example.spanner.SampleRunner.runSample;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.DatabaseId;
+import com.google.cloud.spanner.KeySet;
+import com.google.cloud.spanner.Mutation;
+import com.google.common.collect.ImmutableList;
+import java.util.Arrays;
+import java.util.Collections;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Integration tests for {@link ChangeStreamsTxnExclusionSample}
+ */
+@RunWith(JUnit4.class)
+public class ChangeStreamsTxnExclusionSampleIT extends SampleTestBase {
+
+  private static DatabaseId databaseId;
+
+  @BeforeClass
+  public static void createTestDatabase() throws Exception {
+    final String database = idGenerator.generateDatabaseId();
+    databaseAdminClient
+        .createDatabase(
+            instanceId,
+            database,
+            ImmutableList.of(
+                "CREATE TABLE Singers ("
+                    + "  SingerId   INT64 NOT NULL,"
+                    + "  FirstName  STRING(1024),"
+                    + "  LastName   STRING(1024),"
+                    + "  SingerInfo BYTES(MAX)"
+                    + ") PRIMARY KEY (SingerId)"))
+        .get();
+    databaseId = DatabaseId.of(projectId, instanceId, database);
+  }
+
+  @Before
+  public void insertTestData() {
+    final DatabaseClient client = spanner.getDatabaseClient(databaseId);
+    client.write(Arrays.asList(
+        Mutation.newInsertBuilder("Singers")
+            .set("SingerId")
+            .to(1L)
+            .set("FirstName")
+            .to("first name 1")
+            .set("LastName")
+            .to("last name 1")
+            .build(),
+        Mutation.newInsertBuilder("Singers")
+            .set("SingerId")
+            .to(2L)
+            .set("FirstName")
+            .to("first name 2")
+            .set("LastName")
+            .to("last name 2")
+            .build()
+    ));
+  }
+
+  @After
+  public void removeTestData() {
+    final DatabaseClient client = spanner.getDatabaseClient(databaseId);
+    client.write(Collections.singletonList(Mutation.delete("Singers", KeySet.all())));
+  }
+
+  @Test
+  public void testSetExcludeTxnFromChangeStreamsSampleSample() throws Exception {
+    final DatabaseClient client = spanner.getDatabaseClient(databaseId);
+    String out = runSample(() -> ChangeStreamsTxnExclusionSample.rwTxnExcludedFromChangeStreams(client));
+    assertThat(out).contains("New singer inserted.");
+    assertThat(out).contains("Singer first name updated.");
+
+    out = runSample(() -> ChangeStreamsTxnExclusionSample.writeExcludedFromChangeStreams(client));
+    assertThat(out).contains("New singer inserted.");
+
+    out = runSample(() -> ChangeStreamsTxnExclusionSample.writeAtLeastOnceExcludedFromChangeStreams(client));
+    assertThat(out).contains("New singer inserted.");
+
+    out = runSample(() -> ChangeStreamsTxnExclusionSample.batchWriteAtLeastOnceExcludedFromChangeStreams(client));
+    assertThat(out).contains("have been applied");
+
+    out = runSample(() -> ChangeStreamsTxnExclusionSample.pdmlExcludedFromChangeStreams(client));
+    assertThat(out).contains("Singers deleted.");
+
+    out = runSample(() -> ChangeStreamsTxnExclusionSample.txnManagerExcludedFromChangeStreams(client));
+    assertThat(out).contains("New singer inserted.");
+  }
+}