feat: support multiplexed sessions for RO transactions (googleapis#3141)

* feat(spanner): environment variable to enable multiplexed sessions

* chore: lint fix

* feat(spanner): add comment for gRPC-GCP

* feat(spanner): make strict check in env variable

* feat(spanner): add test

* feat(spanner): add test

* chore(spanner): lint

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

---------

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: 2 people

.github/workflows/ci.yaml

@@ -52,7 +52,7 @@ jobs:
     - run: .kokoro/build.sh
       env:
         JOB_TYPE: test
-        GOOGLE_CLOUD_SPANNER_ENABLE_MULTIPLEXED_SESSIONS: true
+        GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS: true
   units-java8:
     # Building using Java 17 and run the tests with Java 8 runtime
     name: "units (8)"
@@ -91,7 +91,7 @@ jobs:
       - run: .kokoro/build.sh
         env:
           JOB_TYPE: test
-          GOOGLE_CLOUD_SPANNER_ENABLE_MULTIPLEXED_SESSIONS: true
+          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS: true
   windows:
     runs-on: windows-latest
     steps:

.github/workflows/integration-tests-against-emulator-with-multiplexed-session.yaml

@@ -39,4 +39,4 @@ jobs:
       env:
         JOB_TYPE: test
         SPANNER_EMULATOR_HOST: localhost:9010
-        GOOGLE_CLOUD_SPANNER_ENABLE_MULTIPLEXED_SESSIONS: true
+        GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS: true

.kokoro/presubmit/integration-multiplexed-sessions-enabled.cfg

@@ -33,6 +33,6 @@ env_vars: {
 }
 
 env_vars: {
-    key: "GOOGLE_CLOUD_SPANNER_ENABLE_MULTIPLEXED_SESSIONS"
+    key: "GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS"
     value: "true"
-}
+}

README.md

@@ -57,13 +57,13 @@ implementation 'com.google.cloud:google-cloud-spanner'
 If you are using Gradle without BOM, add this to your dependencies:
 
 ```Groovy
-implementation 'com.google.cloud:google-cloud-spanner:6.69.0'
+implementation 'com.google.cloud:google-cloud-spanner:6.70.0'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-spanner" % "6.69.0"
+libraryDependencies += "com.google.cloud" % "google-cloud-spanner" % "6.70.0"
 ```
 <!-- {x-version-update-end} -->
 
@@ -687,7 +687,7 @@ Java is a registered trademark of Oracle and/or its affiliates.
 [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-spanner/java11.html
 [stability-image]: https://img.shields.io/badge/stability-stable-green
 [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-spanner.svg
-[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-spanner/6.69.0
+[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-spanner/6.70.0
 [authentication]: https://github.com/googleapis/google-cloud-java#authentication
 [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes
 [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles

google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionPoolOptions.java

@@ -16,7 +16,6 @@
 
 package com.google.cloud.spanner;
 
-import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.SessionPool.Position;
 import com.google.common.annotations.VisibleForTesting;
@@ -103,11 +102,12 @@ private SessionPoolOptions(Builder builder) {
     this.randomizePositionQPSThreshold = builder.randomizePositionQPSThreshold;
     this.inactiveTransactionRemovalOptions = builder.inactiveTransactionRemovalOptions;
     this.poolMaintainerClock = builder.poolMaintainerClock;
-    // TODO: Remove when multiplexed sessions are guaranteed to be supported.
+    // useMultiplexedSession priority => Environment var > private setter > client default
+    Boolean useMultiplexedSessionFromEnvVariable = getUseMultiplexedSessionFromEnvVariable();
     this.useMultiplexedSession =
-        builder.useMultiplexedSession
-            && !Boolean.parseBoolean(
-                System.getenv("GOOGLE_CLOUD_SPANNER_FORCE_DISABLE_MULTIPLEXED_SESSIONS"));
+        (useMultiplexedSessionFromEnvVariable != null)
+            ? useMultiplexedSessionFromEnvVariable
+            : builder.useMultiplexedSession;
     this.multiplexedSessionMaintenanceDuration = builder.multiplexedSessionMaintenanceDuration;
   }
 
@@ -307,6 +307,22 @@ public boolean getUseMultiplexedSession() {
     return useMultiplexedSession;
   }
 
+  private static Boolean getUseMultiplexedSessionFromEnvVariable() {
+    String useMultiplexedSessionFromEnvVariable =
+        System.getenv("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS");
+    if (useMultiplexedSessionFromEnvVariable != null
+        && useMultiplexedSessionFromEnvVariable.length() > 0) {
+      if ("true".equalsIgnoreCase(useMultiplexedSessionFromEnvVariable)
+          || "false".equalsIgnoreCase(useMultiplexedSessionFromEnvVariable)) {
+        return Boolean.parseBoolean(useMultiplexedSessionFromEnvVariable);
+      } else {
+        throw new IllegalArgumentException(
+            "GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS should be either true or false.");
+      }
+    }
+    return null;
+  }
+
   Duration getMultiplexedSessionMaintenanceDuration() {
     return multiplexedSessionMaintenanceDuration;
   }
@@ -509,7 +525,9 @@ public static class Builder {
      */
     private long randomizePositionQPSThreshold = 0L;
 
-    private boolean useMultiplexedSession = getUseMultiplexedSessionFromEnvVariable();
+    // This field controls the default behavior of session management in Java client.
+    // Set useMultiplexedSession to true to make multiplexed session the default.
+    private boolean useMultiplexedSession = false;
 
     private Duration multiplexedSessionMaintenanceDuration = Duration.ofDays(7);
     private Clock poolMaintainerClock = Clock.INSTANCE;
@@ -527,18 +545,6 @@ private static Position getReleaseToPositionFromSystemProperty() {
       return Position.FIRST;
     }
 
-    /**
-     * This environment is only added to support internal spanner testing. Support for it can be
-     * removed in the future. Use {@link SessionPoolOptions#useMultiplexedSession} instead to use
-     * multiplexed sessions.
-     */
-    @InternalApi
-    @BetaApi
-    private static boolean getUseMultiplexedSessionFromEnvVariable() {
-      return Boolean.parseBoolean(
-          System.getenv("GOOGLE_CLOUD_SPANNER_ENABLE_MULTIPLEXED_SESSIONS"));
-    }
-
     public Builder() {}
 
     private Builder(SessionPoolOptions options) {

google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java

@@ -1287,14 +1287,20 @@ public Builder setHost(String host) {
       return this;
     }
 
-    /** Enables gRPC-GCP extension with the default settings. */
+    /**
+     * Enables gRPC-GCP extension with the default settings. Do not set
+     * GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS to true in combination with this option, as
+     * Multiplexed sessions are not supported for gRPC-GCP.
+     */
     public Builder enableGrpcGcpExtension() {
       return this.enableGrpcGcpExtension(null);
     }
 
     /**
      * Enables gRPC-GCP extension and uses provided options for configuration. The metric registry
-     * and default Spanner metric labels will be added automatically.
+     * and default Spanner metric labels will be added automatically. Do not set
+     * GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS to true in combination with this option, as
+     * Multiplexed sessions are not supported for gRPC-GCP.
      */
     public Builder enableGrpcGcpExtension(GcpManagedChannelOptions options) {
       this.grpcGcpExtensionEnabled = true;

google-cloud-spanner/src/test/java/com/google/cloud/spanner/MultiplexedSessionDatabaseClientTest.java

@@ -16,8 +16,10 @@
 
 package com.google.cloud.spanner;
 
+import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeFalse;
 import static org.junit.Assume.assumeTrue;
@@ -27,6 +29,8 @@
 import static org.mockito.Mockito.when;
 
 import com.google.cloud.spanner.SessionClient.SessionConsumer;
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import java.lang.reflect.Field;
 import java.time.Clock;
 import java.time.Duration;
@@ -110,10 +114,9 @@ public void testMaintainer() {
   }
 
   @Test
-  public void testForceDisableEnvVar() throws Exception {
+  public void testDisableMultiplexedSessionEnvVar() throws Exception {
     assumeTrue(isJava8() && !isWindows());
-    assumeFalse(
-        System.getenv().containsKey("GOOGLE_CLOUD_SPANNER_FORCE_DISABLE_MULTIPLEXED_SESSIONS"));
+    assumeFalse(System.getenv().containsKey("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS"));
 
     // Assert that the mux sessions setting is respected by default.
     assertTrue(
@@ -129,17 +132,116 @@ public void testForceDisableEnvVar() throws Exception {
         (Map<String, String>) field.get(System.getenv());
 
     try {
-      writeableEnvironmentVariables.put(
-          "GOOGLE_CLOUD_SPANNER_FORCE_DISABLE_MULTIPLEXED_SESSIONS", "true");
+      writeableEnvironmentVariables.put("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS", "false");
       // Assert that the env var overrides the mux sessions setting.
       assertFalse(
           SessionPoolOptions.newBuilder()
               .setUseMultiplexedSession(true)
               .build()
               .getUseMultiplexedSession());
     } finally {
-      writeableEnvironmentVariables.remove(
-          "GOOGLE_CLOUD_SPANNER_FORCE_DISABLE_MULTIPLEXED_SESSIONS");
+      writeableEnvironmentVariables.remove("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS");
+    }
+  }
+
+  @Test
+  public void testEnableMultiplexedSessionEnvVar() throws Exception {
+    assumeTrue(isJava8() && !isWindows());
+    assumeFalse(System.getenv().containsKey("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS"));
+
+    // Assert that the mux sessions setting is respected by default.
+    assertFalse(
+        SessionPoolOptions.newBuilder()
+            .setUseMultiplexedSession(false)
+            .build()
+            .getUseMultiplexedSession());
+
+    Class<?> classOfMap = System.getenv().getClass();
+    Field field = classOfMap.getDeclaredField("m");
+    field.setAccessible(true);
+    Map<String, String> writeableEnvironmentVariables =
+        (Map<String, String>) field.get(System.getenv());
+
+    try {
+      writeableEnvironmentVariables.put("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS", "true");
+      // Assert that the env var overrides the mux sessions setting.
+      assertTrue(
+          SessionPoolOptions.newBuilder()
+              .setUseMultiplexedSession(false)
+              .build()
+              .getUseMultiplexedSession());
+    } finally {
+      writeableEnvironmentVariables.remove("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS");
+    }
+  }
+
+  @Test
+  public void testIgnoreMultiplexedSessionEnvVar() throws Exception {
+    assumeTrue(isJava8() && !isWindows());
+    assumeFalse(System.getenv().containsKey("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS"));
+
+    // Assert that the mux sessions setting is respected by default.
+    assertFalse(
+        SessionPoolOptions.newBuilder()
+            .setUseMultiplexedSession(false)
+            .build()
+            .getUseMultiplexedSession());
+
+    Class<?> classOfMap = System.getenv().getClass();
+    Field field = classOfMap.getDeclaredField("m");
+    field.setAccessible(true);
+    Map<String, String> writeableEnvironmentVariables =
+        (Map<String, String>) field.get(System.getenv());
+
+    try {
+      writeableEnvironmentVariables.put("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS", "");
+      // Assert that the env var overrides the mux sessions setting.
+      assertFalse(
+          SessionPoolOptions.newBuilder()
+              .setUseMultiplexedSession(false)
+              .build()
+              .getUseMultiplexedSession());
+    } finally {
+      writeableEnvironmentVariables.remove("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS");
+    }
+  }
+
+  @Test
+  public void testThrowExceptionMultiplexedSessionEnvVarInvalidValues() throws Exception {
+    assumeTrue(isJava8() && !isWindows());
+    assumeFalse(System.getenv().containsKey("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS"));
+
+    // Assert that the mux sessions setting is respected by default.
+    assertFalse(
+        SessionPoolOptions.newBuilder()
+            .setUseMultiplexedSession(false)
+            .build()
+            .getUseMultiplexedSession());
+
+    Class<?> classOfMap = System.getenv().getClass();
+    Field field = classOfMap.getDeclaredField("m");
+    field.setAccessible(true);
+    Map<String, String> writeableEnvironmentVariables =
+        (Map<String, String>) field.get(System.getenv());
+
+    try {
+      writeableEnvironmentVariables.put("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS", "test");
+
+      // setting an invalid GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS value throws error.
+      IllegalArgumentException e =
+          assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  SessionPoolOptions.newBuilder()
+                      .setUseMultiplexedSession(false)
+                      .build()
+                      .getUseMultiplexedSession());
+      StringWriter sw = new StringWriter();
+      e.printStackTrace(new PrintWriter(sw));
+      assertThat(sw.toString())
+          .contains("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS should be either true or false");
+    } finally {
+      writeableEnvironmentVariables.remove("GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS");
     }
   }