chore: allow non-default service account for DP (#2635)

* chore: allow non-default service account for DP

* 🦉 Updates from OwlBot post-processor

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

* fix: only try DP with credentials

* fix: add option for disabling direct path

* fix: disable direct path for test

---------

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

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

@@ -17,6 +17,7 @@
 package com.google.cloud.spanner;
 
 import com.google.api.core.ApiFunction;
+import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.api.gax.core.ExecutorProvider;
 import com.google.api.gax.grpc.GrpcCallContext;
@@ -135,6 +136,7 @@ public class SpannerOptions extends ServiceOptions<Spanner, SpannerOptions> {
   private final CloseableExecutorProvider asyncExecutorProvider;
   private final String compressorName;
   private final boolean leaderAwareRoutingEnabled;
+  private final boolean attemptDirectPath;
 
   /** Interface that can be used to provide {@link CallCredentials} to {@link SpannerOptions}. */
   public interface CallCredentialsProvider {
@@ -624,6 +626,7 @@ private SpannerOptions(Builder builder) {
     asyncExecutorProvider = builder.asyncExecutorProvider;
     compressorName = builder.compressorName;
     leaderAwareRoutingEnabled = builder.leaderAwareRoutingEnabled;
+    attemptDirectPath = builder.attemptDirectPath;
   }
 
   /**
@@ -725,6 +728,7 @@ public static class Builder
     private String compressorName;
     private String emulatorHost = System.getenv("SPANNER_EMULATOR_HOST");
     private boolean leaderAwareRoutingEnabled = true;
+    private boolean attemptDirectPath = true;
 
     private static String createCustomClientLibToken(String token) {
       return token + " " + ServiceOptions.getGoogApiClientLibName();
@@ -784,6 +788,7 @@ private Builder() {
       this.channelProvider = options.channelProvider;
       this.channelConfigurator = options.channelConfigurator;
       this.interceptorProvider = options.interceptorProvider;
+      this.attemptDirectPath = options.attemptDirectPath;
     }
 
     @Override
@@ -1220,6 +1225,12 @@ public Builder disableLeaderAwareRouting() {
       return this;
     }
 
+    @BetaApi
+    public Builder disableDirectPath() {
+      this.attemptDirectPath = false;
+      return this;
+    }
+
     @SuppressWarnings("rawtypes")
     @Override
     public SpannerOptions build() {
@@ -1360,6 +1371,11 @@ public boolean isLeaderAwareRoutingEnabled() {
     return leaderAwareRoutingEnabled;
   }
 
+  @BetaApi
+  public boolean isAttemptDirectPath() {
+    return attemptDirectPath;
+  }
+
   /** Returns the default query options to use for the specific database. */
   public QueryOptions getDefaultQueryOptions(DatabaseId databaseId) {
     // Use the specific query options for the database if any have been specified. These have

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

@@ -53,6 +53,7 @@
 import com.google.api.gax.rpc.UnavailableException;
 import com.google.api.gax.rpc.WatchdogProvider;
 import com.google.api.pathtemplate.PathTemplate;
+import com.google.cloud.NoCredentials;
 import com.google.cloud.RetryHelper;
 import com.google.cloud.RetryHelper.RetryHelperException;
 import com.google.cloud.grpc.GcpManagedChannelBuilder;
@@ -339,9 +340,16 @@ public GapicSpannerRpc(final SpannerOptions options) {
                       // This sets the response compressor (Server -> Client).
                       .withEncoding(compressorName))
               .setHeaderProvider(headerProviderWithUserAgent)
+              .setAllowNonDefaultServiceAccount(true)
               // Attempts direct access to spanner service over gRPC to improve throughput,
               // whether the attempt is allowed is totally controlled by service owner.
-              .setAttemptDirectPath(true);
+              // We'll only attempt DirectPath if we are using real credentials.
+              // NoCredentials is used for plain text connections, for example when connecting to
+              // the emulator.
+              .setAttemptDirectPath(
+                  options.isAttemptDirectPath()
+                      && !Objects.equals(
+                          options.getScopedCredentials(), NoCredentials.getInstance()));
 
       // If it is enabled in options uses the channel pool provided by the gRPC-GCP extension.
       maybeEnableGrpcGcpExtension(defaultChannelProviderBuilder, options);

google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/CredentialsProviderTest.java

@@ -81,9 +81,10 @@ public void testCredentialsProvider() throws Throwable {
                           "cloudspanner://localhost:%d/projects/proj/instances/inst/databases/db?credentialsProvider=%s",
                           getPort(), TestCredentialsProvider.class.getName()))
                   .setConfigurator(
-                      spannerOptions ->
-                          spannerOptions.setChannelConfigurator(
-                              ManagedChannelBuilder::usePlaintext))
+                      spannerOptions -> {
+                        spannerOptions.setChannelConfigurator(ManagedChannelBuilder::usePlaintext);
+                        spannerOptions.disableDirectPath();
+                      })
                   .build();
 
           try (Connection connection = options.getConnection()) {
@@ -122,9 +123,10 @@ public void testCredentialsProvider() throws Throwable {
                           "cloudspanner://localhost:%d/projects/proj/instances/inst/databases/db?credentialsProvider=%s",
                           getPort(), TestCredentialsProvider.class.getName()))
                   .setConfigurator(
-                      spannerOptions ->
-                          spannerOptions.setChannelConfigurator(
-                              ManagedChannelBuilder::usePlaintext))
+                      spannerOptions -> {
+                        spannerOptions.setChannelConfigurator(ManagedChannelBuilder::usePlaintext);
+                        spannerOptions.disableDirectPath();
+                      })
                   .build();
           try (Connection connection = options.getConnection()) {
             assertEquals(

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

@@ -63,6 +63,7 @@
 import com.google.spanner.v1.TypeCode;
 import io.grpc.Context;
 import io.grpc.Contexts;
+import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
 import io.grpc.Metadata.Key;
 import io.grpc.MethodDescriptor;
@@ -641,11 +642,8 @@ private SpannerOptions createSpannerOptions() {
     return SpannerOptions.newBuilder()
         .setProjectId("[PROJECT]")
         // Set a custom channel configurator to allow http instead of https.
-        .setChannelConfigurator(
-            input -> {
-              input.usePlaintext();
-              return input;
-            })
+        .setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
+        .disableDirectPath()
         .setHost("http://" + endpoint)
         // Set static credentials that will return the static OAuth test token.
         .setCredentials(STATIC_CREDENTIALS)

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

@@ -34,6 +34,7 @@
 import com.google.spanner.v1.StructType;
 import com.google.spanner.v1.TypeCode;
 import io.grpc.ForwardingServerCall;
+import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
 import io.grpc.Server;
 import io.grpc.ServerCall;
@@ -311,11 +312,8 @@ private static SpannerOptions createSpannerOptions(InetSocketAddress address, Se
     return SpannerOptions.newBuilder()
         .setProjectId("[PROJECT]")
         // Set a custom channel configurator to allow http instead of https.
-        .setChannelConfigurator(
-            input -> {
-              input.usePlaintext();
-              return input;
-            })
+        .setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
+        .disableDirectPath()
         .setHost("http://" + endpoint)
         // Set static credentials that will return the static OAuth test token.
         .setCredentials(STATIC_CREDENTIALS)