feat(spanner): Cleanup GcpFallbackChannel creation and enable by default (#12707)

* Clean up setupGcpFallback by using new getDecoratedChannelBuilder
instead of capturing cloudpath ChannelConfigurator
* Set EEF period to 3 minutes
* Enable fallback by default (still only used when direct access is
enabled)

Author: kinsaurralde

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

@@ -1041,7 +1041,7 @@ default boolean isEnableDirectAccess() {
     }
 
     default boolean isEnableGcpFallback() {
-      return false;
+      return true;
     }
 
     default boolean isEnableBuiltInMetrics() {
@@ -1136,7 +1136,8 @@ public boolean isEnableDirectAccess() {
 
     @Override
     public boolean isEnableGcpFallback() {
-      return Boolean.parseBoolean(System.getenv(GOOGLE_SPANNER_ENABLE_GCP_FALLBACK));
+      String enableGcpFallback = System.getenv(GOOGLE_SPANNER_ENABLE_GCP_FALLBACK);
+      return enableGcpFallback == null ? true : Boolean.parseBoolean(enableGcpFallback);
     }
 
     @Override

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java

@@ -230,7 +230,6 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
@@ -587,6 +586,7 @@ GcpFallbackChannelOptions createFallbackChannelOptions(
         .setPrimaryChannelName("directpath")
         .setFallbackChannelName("cloudpath")
         .setMinFailedCalls(minFailedCalls)
+        .setPeriod(Duration.ofMinutes(3))
         .setGcpFallbackOpenTelemetry(fallbackTelemetry)
         .build();
   }
@@ -659,27 +659,13 @@ private void setupGcpFallback(
         createChannelProviderBuilder(
             options, headerProviderWithUserAgent, /* isEnableDirectAccess= */ false);
 
-    final ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> existingCloudPathConfigurator =
-        cloudPathProviderBuilder.getChannelConfigurator();
-    final AtomicReference<ManagedChannelBuilder> cloudPathBuilderRef = new AtomicReference<>();
-    cloudPathProviderBuilder.setChannelConfigurator(
-        builder -> {
-          ManagedChannelBuilder effectiveBuilder = builder;
-          if (existingCloudPathConfigurator != null) {
-            effectiveBuilder = existingCloudPathConfigurator.apply(effectiveBuilder);
-          }
-          cloudPathBuilderRef.set(effectiveBuilder);
-          return effectiveBuilder;
-        });
-
-    // Build the cloudPathProvider to extract the builder which will be provided to
-    // FallbackChannelBuilder.
-    try (TransportChannel ignored = cloudPathProviderBuilder.build().getTransportChannel()) {
-    } catch (Exception e) {
+    InstantiatingGrpcChannelProvider cloudPathProvider = cloudPathProviderBuilder.build();
+    ManagedChannelBuilder cloudPathBuilder;
+    try {
+      cloudPathBuilder = cloudPathProvider.createDecoratedChannelBuilder();
+    } catch (IOException e) {
       throw asSpannerException(e);
     }
-
-    ManagedChannelBuilder cloudPathBuilder = cloudPathBuilderRef.get();
     if (cloudPathBuilder == null) {
       throw new IllegalStateException("CloudPath builder was not captured.");
     }

java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java

@@ -1442,61 +1442,49 @@ public void testFallbackIntegration_doesNotSwitchWhenThresholdNotMet() throws Ex
     OpenTelemetrySdk openTelemetry =
         OpenTelemetrySdk.builder().setMeterProvider(meterProvider).build();
 
-    SpannerOptions.useEnvironment(
-        new SpannerOptions.SpannerEnvironment() {
-          @Override
-          public boolean isEnableGcpFallback() {
-            return true;
-          }
-        });
+    SpannerOptions.Builder builder =
+        SpannerOptions.newBuilder()
+            .setProjectId("test-project")
+            .setEnableDirectAccess(true)
+            .setHost("http://localhost:1") // Closed port
+            .setCredentials(NoCredentials.getInstance())
+            .setOpenTelemetry(openTelemetry);
+    // Make sure the ExecuteBatchDml RPC fails quickly to keep the test fast.
+    // Note that the timeout is actually not used. It is the fact that it does not retry that
+    // makes it fail fast.
+    builder
+        .getSpannerStubSettingsBuilder()
+        .executeBatchDmlSettings()
+        .setSimpleTimeoutNoRetriesDuration(Duration.ofSeconds(10));
+    // Setup Options with invalid host to force error
+    SpannerOptions options = builder.build();
+
+    TestableGapicSpannerRpc rpc = new TestableGapicSpannerRpc(options);
     try {
-      SpannerOptions.Builder builder =
-          SpannerOptions.newBuilder()
-              .setProjectId("test-project")
-              .setEnableDirectAccess(true)
-              .setHost("http://localhost:1") // Closed port
-              .setCredentials(NoCredentials.getInstance())
-              .setOpenTelemetry(openTelemetry);
-      // Make sure the ExecuteBatchDml RPC fails quickly to keep the test fast.
-      // Note that the timeout is actually not used. It is the fact that it does not retry that
-      // makes it fail fast.
-      builder
-          .getSpannerStubSettingsBuilder()
-          .executeBatchDmlSettings()
-          .setSimpleTimeoutNoRetriesDuration(Duration.ofSeconds(10));
-      // Setup Options with invalid host to force error
-      SpannerOptions options = builder.build();
-
-      TestableGapicSpannerRpc rpc = new TestableGapicSpannerRpc(options);
-      try {
-        // Make a call that is expected to fail
-        SpannerException exception =
-            assertThrows(
-                SpannerException.class,
-                () ->
-                    rpc.executeBatchDml(
-                        com.google.spanner.v1.ExecuteBatchDmlRequest.newBuilder()
-                            .setSession("projects/p/instances/i/databases/d/sessions/s")
-                            .build(),
-                        null));
-        assertEquals(ErrorCode.UNAVAILABLE, exception.getErrorCode());
-
-        // Wait briefly for the 10ms period to trigger the fallback check
-        Thread.sleep(10);
-
-        // Verify Fallback via Metrics
-        Collection<MetricData> metrics = metricReader.collectAllMetrics();
-        boolean fallbackOccurred =
-            metrics.stream()
-                .anyMatch(md -> md.getName().contains("fallback_count") && hasValue(md));
-
-        assertFalse("Fallback metric should not be present", fallbackOccurred);
+      // Make a call that is expected to fail
+      SpannerException exception =
+          assertThrows(
+              SpannerException.class,
+              () ->
+                  rpc.executeBatchDml(
+                      com.google.spanner.v1.ExecuteBatchDmlRequest.newBuilder()
+                          .setSession("projects/p/instances/i/databases/d/sessions/s")
+                          .build(),
+                      null));
+      assertEquals(ErrorCode.UNAVAILABLE, exception.getErrorCode());
+
+      // Wait briefly for the 10ms period to trigger the fallback check
+      Thread.sleep(10);
+
+      // Verify Fallback via Metrics
+      Collection<MetricData> metrics = metricReader.collectAllMetrics();
+      boolean fallbackOccurred =
+          metrics.stream().anyMatch(md -> md.getName().contains("fallback_count") && hasValue(md));
+
+      assertFalse("Fallback metric should not be present", fallbackOccurred);
 
-      } finally {
-        rpc.shutdown();
-      }
     } finally {
-      SpannerOptions.useDefaultEnvironment();
+      rpc.shutdown();
     }
   }
 
@@ -1533,64 +1521,52 @@ public void testFallbackIntegration_switchesToFallbackOnFailure() throws Excepti
     OpenTelemetrySdk openTelemetry =
         OpenTelemetrySdk.builder().setMeterProvider(meterProvider).build();
 
-    SpannerOptions.useEnvironment(
-        new SpannerOptions.SpannerEnvironment() {
-          @Override
-          public boolean isEnableGcpFallback() {
-            return true;
-          }
-        });
+    SpannerOptions.Builder builder =
+        SpannerOptions.newBuilder()
+            .setProjectId("test-project")
+            .setEnableDirectAccess(true)
+            .setHost("http://localhost:1") // Closed port
+            .setCredentials(NoCredentials.getInstance())
+            .setOpenTelemetry(openTelemetry);
+    // Make sure the ExecuteBatchDml RPC fails quickly to keep the test fast.
+    // Note that the timeout is actually not used. It is the fact that it does not retry that
+    // makes it fail fast.
+    builder
+        .getSpannerStubSettingsBuilder()
+        .executeBatchDmlSettings()
+        .setSimpleTimeoutNoRetriesDuration(Duration.ofSeconds(10));
+    // Setup Options with invalid host to force error
+    SpannerOptions options = builder.build();
+
+    TestableGapicSpannerRpcWithLowerMinFailedCalls rpc =
+        new TestableGapicSpannerRpcWithLowerMinFailedCalls(options);
     try {
-      SpannerOptions.Builder builder =
-          SpannerOptions.newBuilder()
-              .setProjectId("test-project")
-              .setEnableDirectAccess(true)
-              .setHost("http://localhost:1") // Closed port
-              .setCredentials(NoCredentials.getInstance())
-              .setOpenTelemetry(openTelemetry);
-      // Make sure the ExecuteBatchDml RPC fails quickly to keep the test fast.
-      // Note that the timeout is actually not used. It is the fact that it does not retry that
-      // makes it fail fast.
-      builder
-          .getSpannerStubSettingsBuilder()
-          .executeBatchDmlSettings()
-          .setSimpleTimeoutNoRetriesDuration(Duration.ofSeconds(10));
-      // Setup Options with invalid host to force error
-      SpannerOptions options = builder.build();
-
-      TestableGapicSpannerRpcWithLowerMinFailedCalls rpc =
-          new TestableGapicSpannerRpcWithLowerMinFailedCalls(options);
-      try {
-        // Make a call that is expected to fail
-        SpannerException exception =
-            assertThrows(
-                SpannerException.class,
-                () ->
-                    rpc.executeBatchDml(
-                        com.google.spanner.v1.ExecuteBatchDmlRequest.newBuilder()
-                            .setSession("projects/p/instances/i/databases/d/sessions/s")
-                            .build(),
-                        null));
-        assertEquals(ErrorCode.UNAVAILABLE, exception.getErrorCode());
-
-        // Wait briefly for the 10ms period to trigger the fallback check
-        Thread.sleep(10);
-
-        // Verify Fallback via Metrics
-        Collection<MetricData> metrics = metricReader.collectAllMetrics();
-        boolean fallbackOccurred =
-            metrics.stream()
-                .anyMatch(md -> md.getName().contains("fallback_count") && hasValue(md));
-
-        assertTrue(
-            "Fallback metric should be present, indicating GcpFallbackChannel is active",
-            fallbackOccurred);
+      // Make a call that is expected to fail
+      SpannerException exception =
+          assertThrows(
+              SpannerException.class,
+              () ->
+                  rpc.executeBatchDml(
+                      com.google.spanner.v1.ExecuteBatchDmlRequest.newBuilder()
+                          .setSession("projects/p/instances/i/databases/d/sessions/s")
+                          .build(),
+                      null));
+      assertEquals(ErrorCode.UNAVAILABLE, exception.getErrorCode());
+
+      // Wait briefly for the 10ms period to trigger the fallback check
+      Thread.sleep(10);
+
+      // Verify Fallback via Metrics
+      Collection<MetricData> metrics = metricReader.collectAllMetrics();
+      boolean fallbackOccurred =
+          metrics.stream().anyMatch(md -> md.getName().contains("fallback_count") && hasValue(md));
+
+      assertTrue(
+          "Fallback metric should be present, indicating GcpFallbackChannel is active",
+          fallbackOccurred);
 
-      } finally {
-        rpc.shutdown();
-      }
     } finally {
-      SpannerOptions.useDefaultEnvironment();
+      rpc.shutdown();
     }
   }