[core,api,auth: Choose executor based on need for thread affinity (#9504)

* core,api,auth: Choose the callOptions executor when applying request metadata to credentials during newStream based upon whether AppEngineCredentials are being used as they require a specific thread to do the processing.

Add an interface to differentiate whether the specific thread is needed.

Fixes b/244209681

api/src/main/java/io/grpc/InternalMayRequireSpecificExecutor.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2022 The gRPC Authors
+ *
+ * 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 io.grpc;
+
+@Internal
+public interface InternalMayRequireSpecificExecutor {
+  boolean isSpecificExecutorRequired();
+}

auth/src/main/java/io/grpc/auth/GoogleAuthLibraryCallCredentials.java

@@ -22,6 +22,7 @@ import com.google.auth.Credentials;
 import com.google.auth.RequestMetadataCallback;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.io.BaseEncoding;
+import io.grpc.InternalMayRequireSpecificExecutor;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.SecurityLevel;
@@ -44,13 +45,16 @@ import javax.annotation.Nullable;
 /**
  * Wraps {@link Credentials} as a {@link io.grpc.CallCredentials}.
  */
-final class GoogleAuthLibraryCallCredentials extends io.grpc.CallCredentials {
+final class GoogleAuthLibraryCallCredentials extends io.grpc.CallCredentials
+    implements InternalMayRequireSpecificExecutor {
   private static final Logger log
       = Logger.getLogger(GoogleAuthLibraryCallCredentials.class.getName());
   private static final JwtHelper jwtHelper
       = createJwtHelperOrNull(GoogleAuthLibraryCallCredentials.class.getClassLoader());
   private static final Class<? extends Credentials> googleCredentialsClass
       = loadGoogleCredentialsClass();
+  private static final Class<?> appEngineCredentialsClass
+      = loadAppEngineCredentials();
 
   private final boolean requirePrivacy;
   @VisibleForTesting
@@ -59,6 +63,8 @@ final class GoogleAuthLibraryCallCredentials extends io.grpc.CallCredentials {
   private Metadata lastHeaders;
   private Map<String, List<String>> lastMetadata;
 
+  private Boolean requiresSpecificExecutor;
+
   public GoogleAuthLibraryCallCredentials(Credentials creds) {
     this(creds, jwtHelper);
   }
@@ -242,6 +248,16 @@ final class GoogleAuthLibraryCallCredentials extends io.grpc.CallCredentials {
     return rawGoogleCredentialsClass.asSubclass(Credentials.class);
   }
 
+  @Nullable
+  private static Class<?> loadAppEngineCredentials() {
+    try {
+      return Class.forName("com.google.auth.appengine.AppEngineCredentials");
+    } catch (ClassNotFoundException ex) {
+      log.log(Level.FINE, "AppEngineCredentials not available in classloader", ex);
+      return null;
+    }
+  }
+
   private static class MethodPair {
     private final Method getter;
     private final Method builderSetter;
@@ -353,4 +369,24 @@ final class GoogleAuthLibraryCallCredentials extends io.grpc.CallCredentials {
       return creds;
     }
   }
+
+  /**
+   * This method is to support the hack for AppEngineCredentials which need to run on a
+   * specific thread.
+   * @return Whether a specific executor is needed or if any executor can be used
+   */
+  @Override
+  public boolean isSpecificExecutorRequired() {
+    // Cache the value so we only need to try to load the class once
+    if (requiresSpecificExecutor == null) {
+      if (appEngineCredentialsClass == null) {
+        requiresSpecificExecutor = Boolean.FALSE;
+      } else {
+        requiresSpecificExecutor = appEngineCredentialsClass.isInstance(creds);
+      }
+    }
+
+    return requiresSpecificExecutor;
+  }
+
 }

core/src/main/java/io/grpc/internal/CallCredentialsApplyingTransportFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 The gRPC Authors
+ * Copyright 2016,2022 The gRPC Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,6 +27,7 @@ import io.grpc.ChannelCredentials;
 import io.grpc.ChannelLogger;
 import io.grpc.ClientStreamTracer;
 import io.grpc.CompositeCallCredentials;
+import io.grpc.InternalMayRequireSpecificExecutor;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.SecurityLevel;
@@ -144,7 +145,21 @@ final class CallCredentialsApplyingTransportFactory implements ClientTransportFa
             }
           };
         try {
-          creds.applyRequestMetadata(requestInfo, appExecutor, applier);
+          // Hack to allow appengine to work when using AppEngineCredentials (b/244209681)
+          // since processing must happen on a specific thread.
+          //
+          // Ideally would always use appExecutor and we could eliminate the interface
+          // InternalMayRequireSpecificExecutor
+          Executor executor;
+          if (creds instanceof InternalMayRequireSpecificExecutor
+              && ((InternalMayRequireSpecificExecutor)creds).isSpecificExecutorRequired()
+              && callOptions.getExecutor() != null) {
+            executor = callOptions.getExecutor();
+          } else {
+            executor = appExecutor;
+          }
+
+          creds.applyRequestMetadata(requestInfo, executor, applier);
         } catch (Throwable t) {
           applier.fail(Status.UNAUTHENTICATED
               .withDescription("Credentials should use fail() instead of throwing exceptions")