Merge branch 'master' of github.com:grpc/grpc-java into impl/xds_timeout_with_max_stream_duration

Author: voidzcy

alts/src/main/java/io/grpc/alts/internal/AltsAuthContext.java

@@ -34,7 +34,7 @@ public AltsAuthContext(HandshakerResult result) {
             .setPeerServiceAccount(result.getPeerIdentity().getServiceAccount())
             .setLocalServiceAccount(result.getLocalIdentity().getServiceAccount())
             .setPeerRpcVersions(result.getPeerRpcVersions())
-            .putAllPeerAttributes(result.getPeerIdentity().getAttributes())
+            .putAllPeerAttributes(result.getPeerIdentity().getAttributesMap())
             .build();
   }
 

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

@@ -104,7 +104,10 @@ public abstract class ClientCall<ReqT, RespT> {
    * Callbacks for receiving metadata, response messages and completion status from the server.
    *
    * <p>Implementations are free to block for extended periods of time. Implementations are not
-   * required to be thread-safe.
+   * required to be thread-safe, but they must not be thread-hostile. The caller is free to call
+   * an instance from multiple threads, but only one call simultaneously. A single thread may
+   * interleave calls to multiple instances, so implementations using ThreadLocals must be careful
+   * to avoid leaking inappropriate state (e.g., clearing the ThreadLocal before returning).
    */
   public abstract static class Listener<T> {
 

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

@@ -32,6 +32,11 @@
  * <p>Providing authentication credentials is better served by {@link
  * CallCredentials}. But a {@code ClientInterceptor} could set the {@code
  * CallCredentials} within the {@link CallOptions}.
+ *
+ * <p>The interceptor may be called for multiple {@link ClientCall calls} by one or more threads
+ * without completing the previous ones first. Refer to the
+ * {@link io.grpc.ClientCall.Listener ClientCall.Listener} docs for more details regarding thread
+ * safety of the returned listener.
  */
 @ThreadSafe
 public interface ClientInterceptor {

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

@@ -46,7 +46,10 @@ public abstract class ServerCall<ReqT, RespT> {
    * close, which is guaranteed before completion.
    *
    * <p>Implementations are free to block for extended periods of time. Implementations are not
-   * required to be thread-safe.
+   * required to be thread-safe, but they must not be thread-hostile. The caller is free to call
+   * an instance from multiple threads, but only one call simultaneously. A single thread may
+   * interleave calls to multiple instances, so implementations using ThreadLocals must be careful
+   * to avoid leaking inappropriate state (e.g., clearing the ThreadLocal before returning).
    */
   // TODO(ejona86): We need to decide what to do in the case of server closing with non-cancellation
   // before client half closes. It may be that we treat such a case as an error. If we permit such

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

@@ -29,6 +29,11 @@
  * <li>Logging and monitoring call behavior</li>
  * <li>Delegating calls to other servers</li>
  * </ul>
+ *
+ * <p>The interceptor may be called for multiple {@link ServerCall calls} by one or more threads
+ * without completing the previous ones first. Refer to the
+ * {@link io.grpc.ServerCall.Listener ServerCall.Listener} docs for more details regarding thread
+ * safety of the returned listener.
  */
 @ThreadSafe
 public interface ServerInterceptor {

interop-testing/src/main/java/io/grpc/testing/integration/TestServiceClient.java

@@ -61,19 +61,6 @@ public static void main(String[] args) throws Exception {
     client.parseArgs(args);
     client.setUp();
 
-    Runtime.getRuntime().addShutdownHook(new Thread() {
-      @Override
-      @SuppressWarnings("CatchAndPrintStackTrace")
-      public void run() {
-        System.out.println("Shutting down");
-        try {
-          client.tearDown();
-        } catch (Exception e) {
-          e.printStackTrace();
-        }
-      }
-    });
-
     try {
       client.run();
     } finally {

rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java

@@ -61,6 +61,7 @@
 import io.grpc.rls.RlsProtoData.RouteLookupResponse;
 import io.grpc.rls.Throttler.ThrottledException;
 import io.grpc.stub.StreamObserver;
+import io.grpc.util.ForwardingLoadBalancerHelper;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
@@ -110,7 +111,7 @@ final class CachingRlsLbClient {
   private final long staleAgeNanos;
   private final long callTimeoutNanos;
 
-  private final Helper helper;
+  private final RlsLbHelper helper;
   private final ManagedChannel rlsChannel;
   private final RouteLookupServiceStub rlsStub;
   private final RlsPicker rlsPicker;
@@ -119,7 +120,7 @@ final class CachingRlsLbClient {
   private final ChannelLogger logger;
 
   private CachingRlsLbClient(Builder builder) {
-    helper = checkNotNull(builder.helper, "helper");
+    helper = new RlsLbHelper(checkNotNull(builder.helper, "helper"));
     scheduledExecutorService = helper.getScheduledExecutorService();
     synchronizationContext = helper.getSynchronizationContext();
     lbPolicyConfig = checkNotNull(builder.lbPolicyConfig, "lbPolicyConfig");
@@ -200,6 +201,7 @@ public void onError(Throwable t) {
                 logger.log(ChannelLogLevel.DEBUG, "Error looking up route:", t);
                 response.setException(t);
                 throttler.registerBackendResponse(false);
+                helper.propagateRlsError();
               }
 
               @Override
@@ -288,6 +290,41 @@ void requestConnection() {
     rlsChannel.getState(true);
   }
 
+  private static final class RlsLbHelper extends ForwardingLoadBalancerHelper {
+
+    final Helper helper;
+    private ConnectivityState state;
+    private SubchannelPicker picker;
+
+    RlsLbHelper(Helper helper) {
+      this.helper = helper;
+    }
+
+    @Override
+    protected Helper delegate() {
+      return helper;
+    }
+
+    @Override
+    public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
+      state = newState;
+      picker = newPicker;
+      super.updateBalancingState(newState, newPicker);
+    }
+
+    void propagateRlsError() {
+      getSynchronizationContext().execute(new Runnable() {
+        @Override
+        public void run() {
+          if (picker != null) {
+            // Refresh the channel state and let pending RPCs reprocess the picker.
+            updateBalancingState(state, picker);
+          }
+        }
+      });
+    }
+  }
+
   /**
    * Viewer class for cached {@link RouteLookupResponse} and associated {@link ChildPolicyWrapper}.
    */

rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java

@@ -25,7 +25,6 @@
 import static org.mockito.Mockito.CALLS_REAL_METHODS;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
@@ -41,13 +40,16 @@
 import io.grpc.ForwardingChannelBuilder;
 import io.grpc.LoadBalancer;
 import io.grpc.LoadBalancer.Helper;
+import io.grpc.LoadBalancer.PickResult;
 import io.grpc.LoadBalancer.SubchannelPicker;
 import io.grpc.LoadBalancerProvider;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
 import io.grpc.NameResolver;
+import io.grpc.NameResolver.ConfigOrError;
 import io.grpc.Status;
+import io.grpc.Status.Code;
 import io.grpc.SynchronizationContext;
 import io.grpc.inprocess.InProcessChannelBuilder;
 import io.grpc.inprocess.InProcessServerBuilder;
@@ -102,6 +104,7 @@ public class CachingRlsLbClientTest {
 
   private static final RouteLookupConfig ROUTE_LOOKUP_CONFIG = getRouteLookupConfig();
   private static final int SERVER_LATENCY_MILLIS = 10;
+  private static final String DEFAULT_TARGET = "fallback.cloudbigtable.googleapis.com";
 
   @Rule
   public final MockitoRule mocks = MockitoJUnit.rule();
@@ -278,7 +281,9 @@ public void get_updatesLbState() throws Exception {
     rlsServerImpl.setLookupTable(
         ImmutableMap.of(
             routeLookupRequest,
-            new RouteLookupResponse(ImmutableList.of("target"), "header-rls-data-value")));
+            new RouteLookupResponse(
+                ImmutableList.of("primary.cloudbigtable.googleapis.com"),
+                "header-rls-data-value")));
 
     // valid channel
     CachedRouteLookupResponse resp = getInSyncContext(routeLookupRequest);
@@ -298,28 +303,41 @@ public void get_updatesLbState() throws Exception {
     assertThat(stateCaptor.getAllValues())
         .containsExactly(ConnectivityState.CONNECTING, ConnectivityState.READY);
     Metadata headers = new Metadata();
-    pickerCaptor.getValue().pickSubchannel(
+    PickResult pickResult = pickerCaptor.getValue().pickSubchannel(
         new PickSubchannelArgsImpl(
             TestMethodDescriptors.voidMethod().toBuilder().setFullMethodName("foo/bar").build(),
             headers,
             CallOptions.DEFAULT));
+    assertThat(pickResult.getStatus().isOk()).isTrue();
+    assertThat(pickResult.getSubchannel()).isNotNull();
     assertThat(headers.get(RLS_DATA_KEY)).isEqualTo("header-rls-data-value");
 
     // move backoff further back to only test error behavior
     fakeBackoffProvider.nextPolicy = createBackoffPolicy(100, TimeUnit.MILLISECONDS);
     // try to get invalid
     RouteLookupRequest invalidRouteLookupRequest =
         new RouteLookupRequest(
-            "unknown_server", "/doesn/exists", "grpc", ImmutableMap.<String, String>of());
+            "service1", "/doesn/exists", "grpc", ImmutableMap.<String, String>of());
     CachedRouteLookupResponse errorResp = getInSyncContext(invalidRouteLookupRequest);
     assertThat(errorResp.isPending()).isTrue();
     fakeTimeProvider.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS);
 
     errorResp = getInSyncContext(invalidRouteLookupRequest);
     assertThat(errorResp.hasError()).isTrue();
 
-    inOrder.verify(helper, never())
-        .updateBalancingState(any(ConnectivityState.class), any(SubchannelPicker.class));
+    // Channel is still READY because the subchannel for method /foo/bar is still READY.
+    // Method /doesn/exists will use fallback child balancer and fail immediately.
+    inOrder.verify(helper)
+        .updateBalancingState(eq(ConnectivityState.READY), pickerCaptor.capture());
+    pickResult = pickerCaptor.getValue().pickSubchannel(
+        new PickSubchannelArgsImpl(
+            TestMethodDescriptors.voidMethod().toBuilder()
+                .setFullMethodName("doesn/exists")
+                .build(),
+            headers,
+            CallOptions.DEFAULT));
+    assertThat(pickResult.getStatus().getCode()).isEqualTo(Code.UNAVAILABLE);
+    assertThat(pickResult.getStatus().getDescription()).isEqualTo("fallback not available");
   }
 
   @Test
@@ -370,7 +388,7 @@ private static RouteLookupConfig getRouteLookupConfig() {
         /* staleAgeInMillis= */ TimeUnit.SECONDS.toMillis(240),
         /* cacheSizeBytes= */ 1000,
         /* validTargets= */ ImmutableList.of("a valid target"),
-        /* defaultTarget= */ "us_east_1.cloudbigtable.googleapis.com");
+        DEFAULT_TARGET);
   }
 
   private static BackoffPolicy createBackoffPolicy(final long delay, final TimeUnit unit) {
@@ -395,6 +413,10 @@ public BackoffPolicy get() {
     }
   }
 
+  /**
+   * A load balancer that immediately goes to READY when using the rls response target and
+   * immediately fails when using the fallback target.
+   */
   private static final class TestLoadBalancerProvider extends LoadBalancerProvider {
 
     @Override
@@ -412,21 +434,39 @@ public String getPolicyName() {
       return null;
     }
 
+    @Override
+    public ConfigOrError parseLoadBalancingPolicyConfig(
+        Map<String, ?> rawLoadBalancingPolicyConfig) {
+      return ConfigOrError.fromConfig(rawLoadBalancingPolicyConfig);
+    }
+
     @Override
     public LoadBalancer newLoadBalancer(final Helper helper) {
       return new LoadBalancer() {
 
         @Override
         public void handleResolvedAddresses(ResolvedAddresses resolvedAddresses) {
-          // TODO: make the picker accessible
-          helper.updateBalancingState(
-              ConnectivityState.READY,
-              new SubchannelPicker() {
-                @Override
-                public PickResult pickSubchannel(PickSubchannelArgs args) {
-                  return PickResult.withSubchannel(mock(Subchannel.class));
-                }
-              });
+          Map<?, ?> config = (Map<?, ?>) resolvedAddresses.getLoadBalancingPolicyConfig();
+          if (DEFAULT_TARGET.equals(config.get("target"))) {
+            helper.updateBalancingState(
+                ConnectivityState.TRANSIENT_FAILURE,
+                new SubchannelPicker() {
+                  @Override
+                  public PickResult pickSubchannel(PickSubchannelArgs args) {
+                    return PickResult.withError(
+                        Status.UNAVAILABLE.withDescription("fallback not available"));
+                  }
+                });
+          } else {
+            helper.updateBalancingState(
+                ConnectivityState.READY,
+                new SubchannelPicker() {
+                  @Override
+                  public PickResult pickSubchannel(PickSubchannelArgs args) {
+                    return PickResult.withSubchannel(mock(Subchannel.class));
+                  }
+                });
+          }
         }
 
         @Override

xds/src/main/java/io/grpc/xds/EnvoyProtoData.java

@@ -918,16 +918,6 @@ RouteAction getRouteAction() {
       return routeAction;
     }
 
-    // TODO(chengyuanzhang): delete and do not use after routing feature is always ON.
-    boolean isDefaultRoute() {
-      // For backward compatibility, all the other matchers are ignored.
-      String prefix = routeMatch.getPathMatch().getPrefix();
-      if (prefix != null) {
-        return prefix.isEmpty() || prefix.equals("/");
-      }
-      return false;
-    }
-
     @Override
     public boolean equals(Object o) {
       if (this == o) {
@@ -992,17 +982,12 @@ static StructOrError<Route> fromEnvoyProtoRoute(
     }
 
     @VisibleForTesting
-    @SuppressWarnings("deprecation")
     @Nullable
     static StructOrError<RouteMatch> convertEnvoyProtoRouteMatch(
         io.envoyproxy.envoy.config.route.v3.RouteMatch proto) {
       if (proto.getQueryParametersCount() != 0) {
         return null;
       }
-      if (proto.hasCaseSensitive() && !proto.getCaseSensitive().getValue()) {
-        return StructOrError.fromError("Unsupported match option: case insensitive");
-      }
-
       StructOrError<PathMatcher> pathMatch = convertEnvoyProtoPathMatcher(proto);
       if (pathMatch.getErrorDetail() != null) {
         return StructOrError.fromError(pathMatch.getErrorDetail());
@@ -1032,32 +1017,28 @@ static StructOrError<RouteMatch> convertEnvoyProtoRouteMatch(
               pathMatch.getStruct(), Collections.unmodifiableList(headerMatchers), fractionMatch));
     }
 
-    @SuppressWarnings("deprecation")
     private static StructOrError<PathMatcher> convertEnvoyProtoPathMatcher(
         io.envoyproxy.envoy.config.route.v3.RouteMatch proto) {
-      String path = null;
-      String prefix = null;
-      Pattern safeRegEx = null;
+      boolean caseSensitive = proto.getCaseSensitive().getValue();
       switch (proto.getPathSpecifierCase()) {
         case PREFIX:
-          prefix = proto.getPrefix();
-          break;
+          return StructOrError.fromStruct(
+              PathMatcher.fromPrefix(proto.getPrefix(), caseSensitive));
         case PATH:
-          path = proto.getPath();
-          break;
+          return StructOrError.fromStruct(PathMatcher.fromPath(proto.getPath(), caseSensitive));
         case SAFE_REGEX:
           String rawPattern = proto.getSafeRegex().getRegex();
+          Pattern safeRegEx;
           try {
             safeRegEx = Pattern.compile(rawPattern);
           } catch (PatternSyntaxException e) {
             return StructOrError.fromError("Malformed safe regex pattern: " + e.getMessage());
           }
-          break;
+          return StructOrError.fromStruct(PathMatcher.fromRegEx(safeRegEx));
         case PATHSPECIFIER_NOT_SET:
         default:
           return StructOrError.fromError("Unknown path match type");
       }
-      return StructOrError.fromStruct(new PathMatcher(path, prefix, safeRegEx));
     }
 
     private static StructOrError<FractionMatcher> convertEnvoyProtoFraction(