Merge branch 'grpc:master' into master

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

@@ -195,7 +195,10 @@ public void run() {
             }
           }
           if (retryFuture != null) {
-            retryFuture.cancel(false);
+            boolean cancelled = retryFuture.cancel(false);
+            if (cancelled) {
+              inFlightSubStreams.decrementAndGet();
+            }
           }
           if (hedgingFuture != null) {
             hedgingFuture.cancel(false);

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

@@ -33,7 +33,6 @@
 import com.google.protobuf.Any;
 import io.grpc.Context;
 import io.grpc.InternalLogId;
-import io.grpc.LoadBalancerRegistry;
 import io.grpc.Status;
 import io.grpc.SynchronizationContext;
 import io.grpc.SynchronizationContext.ScheduledHandle;
@@ -88,9 +87,7 @@ public void uncaughtException(Thread t, Throwable e) {
           throw new AssertionError(e);
         }
       });
-  private final FilterRegistry filterRegistry = FilterRegistry.getDefaultRegistry();
-  private final LoadBalancerRegistry loadBalancerRegistry
-      = LoadBalancerRegistry.getDefaultRegistry();
+
   private final Map<ServerInfo, ControlPlaneClient> serverChannelMap = new HashMap<>();
   private final Map<XdsResourceType<? extends ResourceUpdate>,
       Map<String, ResourceSubscriber<? extends ResourceUpdate>>>
@@ -174,8 +171,7 @@ public void handleResourceResponse(
       toParseResourceNames = resourceSubscribers.get(xdsResourceType).keySet();
     }
     XdsResourceType.Args args = new XdsResourceType.Args(serverInfo, versionInfo, nonce,
-        bootstrapInfo, filterRegistry, loadBalancerRegistry, tlsContextManager,
-        toParseResourceNames);
+        bootstrapInfo, tlsContextManager, toParseResourceNames);
     handleResourceUpdate(args, resources, xdsResourceType, processingTracker);
   }
 

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

@@ -56,6 +56,8 @@ class XdsClusterResource extends XdsResourceType<CdsUpdate> {
       "type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext";
   private static final String TYPE_URL_UPSTREAM_TLS_CONTEXT_V2 =
       "type.googleapis.com/envoy.api.v2.auth.UpstreamTlsContext";
+  private final LoadBalancerRegistry loadBalancerRegistry
+      = LoadBalancerRegistry.getDefaultRegistry();
 
   private static final XdsClusterResource instance = new XdsClusterResource();
 
@@ -103,7 +105,7 @@ protected CdsUpdate doParse(Args args, Message unpackedMessage) throws ResourceI
       certProviderInstances = args.bootstrapInfo.certProviders().keySet();
     }
     return processCluster((Cluster) unpackedMessage, certProviderInstances,
-        args.serverInfo, args.loadBalancerRegistry);
+        args.serverInfo, loadBalancerRegistry);
   }
 
   @VisibleForTesting

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

@@ -102,15 +102,13 @@ protected LdsUpdate doParse(Args args, Message unpackedMessage)
     Listener listener = (Listener) unpackedMessage;
 
     if (listener.hasApiListener()) {
-      return processClientSideListener(
-          listener, args);
+      return processClientSideListener(listener);
     } else {
-      return processServerSideListener(
-          listener, args);
+      return processServerSideListener(listener, args);
     }
   }
 
-  private LdsUpdate processClientSideListener(Listener listener, Args args)
+  private LdsUpdate processClientSideListener(Listener listener)
       throws ResourceInvalidException {
     // Unpack HttpConnectionManager from the Listener.
     HttpConnectionManager hcm;
@@ -123,7 +121,7 @@ private LdsUpdate processClientSideListener(Listener listener, Args args)
           "Could not parse HttpConnectionManager config from ApiListener", e);
     }
     return LdsUpdate.forApiListener(parseHttpConnectionManager(
-        hcm, args.filterRegistry, true /* isForClient */));
+        hcm, filterRegistry, true /* isForClient */));
   }
 
   private LdsUpdate processServerSideListener(Listener proto, Args args)
@@ -133,7 +131,7 @@ private LdsUpdate processServerSideListener(Listener proto, Args args)
       certProviderInstances = args.bootstrapInfo.certProviders().keySet();
     }
     return LdsUpdate.forTcpListener(parseServerSideListener(proto, args.tlsContextManager,
-        args.filterRegistry, certProviderInstances));
+        filterRegistry, certProviderInstances));
   }
 
   @VisibleForTesting

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

@@ -17,8 +17,6 @@
 package io.grpc.xds;
 
 import static com.google.common.base.Preconditions.checkNotNull;
-import static io.grpc.xds.Bootstrapper.ServerInfo;
-import static io.grpc.xds.XdsClient.ResourceUpdate;
 import static io.grpc.xds.XdsClient.canonifyResourceName;
 import static io.grpc.xds.XdsClient.isResourceNameValid;
 
@@ -29,7 +27,6 @@
 import com.google.protobuf.Message;
 import io.envoyproxy.envoy.service.discovery.v3.Resource;
 import io.grpc.ExperimentalApi;
-import io.grpc.LoadBalancerRegistry;
 import io.grpc.xds.Bootstrapper.ServerInfo;
 import io.grpc.xds.XdsClient.ResourceUpdate;
 import java.util.ArrayList;
@@ -63,6 +60,8 @@ public abstract class XdsResourceType<T extends ResourceUpdate> {
   @VisibleForTesting
   static boolean enablePickFirst = getFlag("GRPC_EXPERIMENTAL_PICKFIRST_LB_CONFIG", true);
 
+  protected final FilterRegistry filterRegistry = FilterRegistry.getDefaultRegistry();
+
   static final String TYPE_URL_CLUSTER_CONFIG =
       "type.googleapis.com/envoy.extensions.clusters.aggregate.v3.ClusterConfig";
   static final String TYPE_URL_TYPED_STRUCT_UDPA =
@@ -91,8 +90,6 @@ public static class Args {
     final String versionInfo;
     final String nonce;
     final Bootstrapper.BootstrapInfo bootstrapInfo;
-    final FilterRegistry filterRegistry;
-    final LoadBalancerRegistry loadBalancerRegistry;
     final TlsContextManager tlsContextManager;
     // Management server is required to always send newly requested resources, even if they
     // may have been sent previously (proactively). Thus, client does not need to cache
@@ -102,16 +99,12 @@ public static class Args {
 
     public Args(ServerInfo serverInfo, String versionInfo, String nonce,
                 Bootstrapper.BootstrapInfo bootstrapInfo,
-                FilterRegistry filterRegistry,
-                LoadBalancerRegistry loadBalancerRegistry,
                 TlsContextManager tlsContextManager,
                 @Nullable Set<String> subscribedResources) {
       this.serverInfo = serverInfo;
       this.versionInfo = versionInfo;
       this.nonce = nonce;
       this.bootstrapInfo = bootstrapInfo;
-      this.filterRegistry = filterRegistry;
-      this.loadBalancerRegistry = loadBalancerRegistry;
       this.tlsContextManager = tlsContextManager;
       this.subscribedResources = subscribedResources;
     }

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

@@ -118,8 +118,7 @@ protected RdsUpdate doParse(XdsResourceType.Args args, Message unpackedMessage)
     if (!(unpackedMessage instanceof RouteConfiguration)) {
       throw new ResourceInvalidException("Invalid message type: " + unpackedMessage.getClass());
     }
-    return processRouteConfiguration((RouteConfiguration) unpackedMessage,
-        args.filterRegistry);
+    return processRouteConfiguration((RouteConfiguration) unpackedMessage, filterRegistry);
   }
 
   private static RdsUpdate processRouteConfiguration(