(Diff) compare changes for CDS only in XdsClient

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

@@ -30,6 +30,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -412,6 +413,7 @@ public static final class Args {
     private final ProxyDetector proxyDetector;
     private final SynchronizationContext syncContext;
     private final ServiceConfigParser serviceConfigParser;
+    @Nullable private final ScheduledExecutorService scheduledExecutorService;
     @Nullable private final ChannelLogger channelLogger;
     @Nullable private final Executor executor;
 
@@ -420,12 +422,14 @@ private Args(
         ProxyDetector proxyDetector,
         SynchronizationContext syncContext,
         ServiceConfigParser serviceConfigParser,
+        @Nullable ScheduledExecutorService scheduledExecutorService,
         @Nullable ChannelLogger channelLogger,
         @Nullable Executor executor) {
       this.defaultPort = checkNotNull(defaultPort, "defaultPort not set");
       this.proxyDetector = checkNotNull(proxyDetector, "proxyDetector not set");
       this.syncContext = checkNotNull(syncContext, "syncContext not set");
       this.serviceConfigParser = checkNotNull(serviceConfigParser, "serviceConfigParser not set");
+      this.scheduledExecutorService = scheduledExecutorService;
       this.channelLogger = channelLogger;
       this.executor = executor;
     }
@@ -460,6 +464,25 @@ public SynchronizationContext getSynchronizationContext() {
       return syncContext;
     }
 
+    /**
+     * Returns a {@link ScheduledExecutorService} for scheduling delayed tasks.
+     *
+     * <p>This service is a shared resource and is only meant for quick tasks. DO NOT block or run
+     * time-consuming tasks.
+     *
+     * <p>The returned service doesn't support {@link ScheduledExecutorService#shutdown shutdown()}
+     *  and {@link ScheduledExecutorService#shutdownNow shutdownNow()}. They will throw if called.
+     *
+     * @since 1.26.0
+     */
+    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/6454")
+    public ScheduledExecutorService getScheduledExecutorService() {
+      if (scheduledExecutorService == null) {
+        throw new IllegalStateException("ScheduledExecutorService not set in Builder");
+      }
+      return scheduledExecutorService;
+    }
+
     /**
      * Returns the {@link ServiceConfigParser}.
      *
@@ -501,6 +524,7 @@ public String toString() {
           .add("proxyDetector", proxyDetector)
           .add("syncContext", syncContext)
           .add("serviceConfigParser", serviceConfigParser)
+          .add("scheduledExecutorService", scheduledExecutorService)
           .add("channelLogger", channelLogger)
           .add("executor", executor)
           .toString();
@@ -517,6 +541,7 @@ public Builder toBuilder() {
       builder.setProxyDetector(proxyDetector);
       builder.setSynchronizationContext(syncContext);
       builder.setServiceConfigParser(serviceConfigParser);
+      builder.setScheduledExecutorService(scheduledExecutorService);
       builder.setChannelLogger(channelLogger);
       builder.setOffloadExecutor(executor);
       return builder;
@@ -541,6 +566,7 @@ public static final class Builder {
       private ProxyDetector proxyDetector;
       private SynchronizationContext syncContext;
       private ServiceConfigParser serviceConfigParser;
+      private ScheduledExecutorService scheduledExecutorService;
       private ChannelLogger channelLogger;
       private Executor executor;
 
@@ -577,6 +603,16 @@ public Builder setSynchronizationContext(SynchronizationContext syncContext) {
         return this;
       }
 
+      /**
+       * See {@link Args#getScheduledExecutorService}.
+       */
+      @ExperimentalApi("https://github.com/grpc/grpc-java/issues/6454")
+      public Builder setScheduledExecutorService(
+          ScheduledExecutorService scheduledExecutorService) {
+        this.scheduledExecutorService = checkNotNull(scheduledExecutorService);
+        return this;
+      }
+
       /**
        * See {@link Args#getServiceConfigParser}.  This is a required field.
        *
@@ -618,7 +654,7 @@ public Args build() {
         return
             new Args(
                 defaultPort, proxyDetector, syncContext, serviceConfigParser,
-                channelLogger, executor);
+                scheduledExecutorService, channelLogger, executor);
       }
     }
   }

api/src/test/java/io/grpc/NameResolverTest.java

@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Before;
 import org.junit.Test;
@@ -45,6 +46,8 @@ public class NameResolverTest {
   private final SynchronizationContext syncContext =
       new SynchronizationContext(mock(UncaughtExceptionHandler.class));
   private final ServiceConfigParser parser = mock(ServiceConfigParser.class);
+  private final ScheduledExecutorService scheduledExecutorService =
+      mock(ScheduledExecutorService.class);
   private final ChannelLogger channelLogger = mock(ChannelLogger.class);
   private final Executor executor = Executors.newSingleThreadExecutor();
   private URI uri;
@@ -62,6 +65,7 @@ public void args() {
     assertThat(args.getProxyDetector()).isSameInstanceAs(proxyDetector);
     assertThat(args.getSynchronizationContext()).isSameInstanceAs(syncContext);
     assertThat(args.getServiceConfigParser()).isSameInstanceAs(parser);
+    assertThat(args.getScheduledExecutorService()).isSameInstanceAs(scheduledExecutorService);
     assertThat(args.getChannelLogger()).isSameInstanceAs(channelLogger);
     assertThat(args.getOffloadExecutor()).isSameInstanceAs(executor);
 
@@ -70,6 +74,7 @@ public void args() {
     assertThat(args2.getProxyDetector()).isSameInstanceAs(proxyDetector);
     assertThat(args2.getSynchronizationContext()).isSameInstanceAs(syncContext);
     assertThat(args2.getServiceConfigParser()).isSameInstanceAs(parser);
+    assertThat(args2.getScheduledExecutorService()).isSameInstanceAs(scheduledExecutorService);
     assertThat(args2.getChannelLogger()).isSameInstanceAs(channelLogger);
     assertThat(args2.getOffloadExecutor()).isSameInstanceAs(executor);
 
@@ -254,6 +259,7 @@ private NameResolver.Args createArgs() {
         .setProxyDetector(proxyDetector)
         .setSynchronizationContext(syncContext)
         .setServiceConfigParser(parser)
+        .setScheduledExecutorService(scheduledExecutorService)
         .setChannelLogger(channelLogger)
         .setOffloadExecutor(executor)
         .build();

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

@@ -137,7 +137,7 @@ final class ManagedChannelImpl extends ManagedChannel implements
   private final NameResolver.Args nameResolverArgs;
   private final AutoConfiguredLoadBalancerFactory loadBalancerFactory;
   private final ClientTransportFactory transportFactory;
-  private final ScheduledExecutorForBalancer scheduledExecutorForBalancer;
+  private final RestrictedScheduledExecutor scheduledExecutor;
   private final Executor executor;
   private final ObjectPool<? extends Executor> executorPool;
   private final ObjectPool<? extends Executor> balancerRpcExecutorPool;
@@ -562,6 +562,12 @@ ClientStream newSubstream(ClientStreamTracer.Factory tracerFactory, Metadata new
     this.target = checkNotNull(builder.target, "target");
     this.logId = InternalLogId.allocate("Channel", target);
     this.timeProvider = checkNotNull(timeProvider, "timeProvider");
+    this.executorPool = checkNotNull(builder.executorPool, "executorPool");
+    this.executor = checkNotNull(executorPool.getObject(), "executor");
+    this.transportFactory =
+        new CallCredentialsApplyingTransportFactory(clientTransportFactory, this.executor);
+    this.scheduledExecutor =
+        new RestrictedScheduledExecutor(transportFactory.getScheduledExecutorService());
     maxTraceEvents = builder.maxTraceEvents;
     channelTracer = new ChannelTracer(
         logId, builder.maxTraceEvents, timeProvider.currentTimeNanos(),
@@ -581,6 +587,7 @@ ClientStream newSubstream(ClientStreamTracer.Factory tracerFactory, Metadata new
             .setDefaultPort(builder.getDefaultPort())
             .setProxyDetector(proxyDetector)
             .setSynchronizationContext(syncContext)
+            .setScheduledExecutorService(scheduledExecutor)
             .setServiceConfigParser(
                 new ScParser(
                     retryEnabled,
@@ -598,18 +605,11 @@ public void execute(Runnable command) {
                 })
             .build();
     this.nameResolver = getNameResolver(target, nameResolverFactory, nameResolverArgs);
-    this.executorPool = checkNotNull(builder.executorPool, "executorPool");
     this.balancerRpcExecutorPool = checkNotNull(balancerRpcExecutorPool, "balancerRpcExecutorPool");
     this.balancerRpcExecutorHolder = new ExecutorHolder(balancerRpcExecutorPool);
-    this.executor = checkNotNull(executorPool.getObject(), "executor");
     this.delayedTransport = new DelayedClientTransport(this.executor, this.syncContext);
     this.delayedTransport.start(delayedTransportListener);
     this.backoffPolicyProvider = backoffPolicyProvider;
-    this.transportFactory =
-        new CallCredentialsApplyingTransportFactory(clientTransportFactory, this.executor);
-    this.scheduledExecutorForBalancer =
-        new ScheduledExecutorForBalancer(transportFactory.getScheduledExecutorService());
-
     serviceConfigInterceptor = new ServiceConfigInterceptor(
         retryEnabled, builder.maxRetryAttempts, builder.maxHedgedAttempts);
     this.defaultServiceConfig = builder.defaultServiceConfig;
@@ -1269,7 +1269,7 @@ public SynchronizationContext getSynchronizationContext() {
 
     @Override
     public ScheduledExecutorService getScheduledExecutorService() {
-      return scheduledExecutorForBalancer;
+      return scheduledExecutor;
     }
 
     @Override
@@ -1736,10 +1736,10 @@ synchronized void release() {
     }
   }
 
-  private static final class ScheduledExecutorForBalancer implements ScheduledExecutorService {
+  private static final class RestrictedScheduledExecutor implements ScheduledExecutorService {
     final ScheduledExecutorService delegate;
 
-    private ScheduledExecutorForBalancer(ScheduledExecutorService delegate) {
+    private RestrictedScheduledExecutor(ScheduledExecutorService delegate) {
       this.delegate = checkNotNull(delegate, "delegate");
     }
 

core/src/test/java/io/grpc/internal/ReadableBuffersArrayTest.java

@@ -23,11 +23,14 @@
 import static org.junit.Assert.assertTrue;
 
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
 
 /**
  * Tests for the array-backed {@link ReadableBuffer} returned by {@link ReadableBuffers#wrap(byte[],
  * int, int)}.
  */
+@RunWith(JUnit4.class)
 public class ReadableBuffersArrayTest extends ReadableBufferTestBase {
 
   @Test

cronet/build.gradle

@@ -42,8 +42,8 @@ android {
     buildTypes {
         debug { minifyEnabled false }
         release {
-            minifyEnabled true
-            proguardFiles getDefaultProguardFile('proguard-android.txt'), 'proguard-rules.pro'
+            minifyEnabled false
+            consumerProguardFiles 'proguard-rules.pro'
         }
     }
     testOptions { unitTests { includeAndroidResources = true } }

netty/src/main/java/io/grpc/netty/Utils.java

@@ -91,7 +91,7 @@ class Utils {
         @Override
         public ByteBufAllocator create() {
           if (Boolean.parseBoolean(
-                  System.getProperty("io.grpc.netty.useCustomAllocator", "false"))) {
+                  System.getProperty("io.grpc.netty.useCustomAllocator", "true"))) {
             int maxOrder;
             if (System.getProperty("io.netty.allocator.maxOrder") == null) {
               // See the implementation of PooledByteBufAllocator.  DEFAULT_MAX_ORDER in there is

stub/src/main/java/io/grpc/stub/ClientCalls.java

@@ -366,15 +366,17 @@ public boolean isReady() {
     @Override
     public void setOnReadyHandler(Runnable onReadyHandler) {
       if (frozen) {
-        throw new IllegalStateException("Cannot alter onReadyHandler after call started");
+        throw new IllegalStateException(
+            "Cannot alter onReadyHandler after call started. Use ClientResponseObserver");
       }
       this.onReadyHandler = onReadyHandler;
     }
 
     @Override
     public void disableAutoInboundFlowControl() {
       if (frozen) {
-        throw new IllegalStateException("Cannot disable auto flow control call started");
+        throw new IllegalStateException(
+            "Cannot disable auto flow control after call started. Use ClientResponseObserver");
       }
       autoFlowControlEnabled = false;
     }

stub/src/main/java/io/grpc/stub/ServerCalls.java

@@ -380,7 +380,9 @@ public boolean isReady() {
 
     @Override
     public void setOnReadyHandler(Runnable r) {
-      checkState(!frozen, "Cannot alter onReadyHandler after initialization");
+      checkState(!frozen, "Cannot alter onReadyHandler after initialization. May only be called "
+          + "during the initial call to the application, before the service returns its "
+          + "StreamObserver");
       this.onReadyHandler = r;
     }
 
@@ -391,7 +393,9 @@ public boolean isCancelled() {
 
     @Override
     public void setOnCancelHandler(Runnable onCancelHandler) {
-      checkState(!frozen, "Cannot alter onCancelHandler after initialization");
+      checkState(!frozen, "Cannot alter onCancelHandler after initialization. May only be called "
+          + "during the initial call to the application, before the service returns its "
+          + "StreamObserver");
       this.onCancelHandler = onCancelHandler;
     }
 

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

@@ -195,6 +195,14 @@ static final class LbEndpoint {
     private final int loadBalancingWeight;
     private final boolean isHealthy;
 
+    @VisibleForTesting
+    LbEndpoint(String address, int port, int loadBalancingWeight, boolean isHealthy) {
+      this(
+          new EquivalentAddressGroup(
+              new InetSocketAddress(address, port)),
+          loadBalancingWeight, isHealthy);
+    }
+
     @VisibleForTesting
     LbEndpoint(EquivalentAddressGroup eag, int loadBalancingWeight, boolean isHealthy) {
       this.eag = eag;

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

@@ -76,23 +76,23 @@ interface LocalityStore {
 
   void updateOobMetricsReportInterval(long reportIntervalNano);
 
-  LoadStatsStore getLoadStatsStore();
-
   @VisibleForTesting
   abstract class LocalityStoreFactory {
     private static final LocalityStoreFactory DEFAULT_INSTANCE =
         new LocalityStoreFactory() {
           @Override
-          LocalityStore newLocalityStore(Helper helper, LoadBalancerRegistry lbRegistry) {
-            return new LocalityStoreImpl(helper, lbRegistry);
+          LocalityStore newLocalityStore(
+              Helper helper, LoadBalancerRegistry lbRegistry, LoadStatsStore loadStatsStore) {
+            return new LocalityStoreImpl(helper, lbRegistry, loadStatsStore);
           }
         };
 
     static LocalityStoreFactory getInstance() {
       return DEFAULT_INSTANCE;
     }
 
-    abstract LocalityStore newLocalityStore(Helper helper, LoadBalancerRegistry lbRegistry);
+    abstract LocalityStore newLocalityStore(
+        Helper helper, LoadBalancerRegistry lbRegistry, LoadStatsStore loadStatsStore);
   }
 
   final class LocalityStoreImpl implements LocalityStore {
@@ -113,9 +113,10 @@ final class LocalityStoreImpl implements LocalityStore {
     private List<DropOverload> dropOverloads = ImmutableList.of();
     private long metricsReportIntervalNano = -1;
 
-    LocalityStoreImpl(Helper helper, LoadBalancerRegistry lbRegistry) {
+    LocalityStoreImpl(
+        Helper helper, LoadBalancerRegistry lbRegistry, LoadStatsStore loadStatsStore) {
       this(helper, pickerFactoryImpl, lbRegistry, ThreadSafeRandom.ThreadSafeRandomImpl.instance,
-          new LoadStatsStoreImpl(), OrcaPerRequestUtil.getInstance(), OrcaOobUtil.getInstance());
+          loadStatsStore, OrcaPerRequestUtil.getInstance(), OrcaOobUtil.getInstance());
     }
 
     @VisibleForTesting
@@ -284,11 +285,6 @@ public String toString() {
           TimeUnit.MINUTES, helper.getScheduledExecutorService());
     }
 
-    @Override
-    public LoadStatsStore getLoadStatsStore() {
-      return loadStatsStore;
-    }
-
     @Override
     public void updateOobMetricsReportInterval(long reportIntervalNano) {
       metricsReportIntervalNano = reportIntervalNano;

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

@@ -175,11 +175,13 @@ public void handleResolvedAddresses(ResolvedAddresses resolvedAddresses) {
                 xdsClient  = new XdsComms2(
                     channel, helper, new ExponentialBackoffPolicy.Provider(),
                     GrpcUtil.STOPWATCH_SUPPLIER, node);
-                localityStore = localityStoreFactory.newLocalityStore(helper, lbRegistry);
+                LoadStatsStore loadStatsStore = new LoadStatsStoreImpl();
+                localityStore = localityStoreFactory.newLocalityStore(
+                    helper, lbRegistry, loadStatsStore);
                 // TODO(zdapeng): Use XdsClient to do Lrs directly.
                 lrsClient = loadReportClientFactory.createLoadReportClient(
                     channel, helper, new ExponentialBackoffPolicy.Provider(),
-                    localityStore.getLoadStatsStore());
+                    loadStatsStore);
                 final LoadReportCallback lrsCallback =
                     new LoadReportCallback() {
                       @Override