Remove gRPC stub cache

src/consumer/Consumer.java

@@ -44,7 +44,7 @@
 public class Consumer extends AbstractService {
 
   /** Version of the client library. Should be kept in sync with Go client version. */
-  public static final String CLIENT_VERSION = "1.2.6";
+  public static final String CLIENT_VERSION = "1.2.7";
 
   private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
 

src/consumer/processor/GrpcRequestForwarder.java

@@ -43,30 +43,6 @@ public interface RemoteHandler<REQ, RESP, STUB extends AbstractBlockingStub<STUB
   private final OwnershipRetryer retryer;
   private final Duration timeout;
 
-  /**
-   * Creates a new forwarder.
-   *
-   * <p>The supplied cache retains its own channel factory and expiration policy;
-   * its channels are not managed by the consumer's peer channel pool.
-   *
-   * @param consumer Splitter consumer
-   * @param grantManager Grant manager, usually an instance of a subclass of {@link
-   *     BaseWorkProcessor}
-   * @param stubsCache Cache with stubs.
-   * @param remoteHandler Invokes gRPC request
-   * @param retryer Retryer to use when forwarding requests
-   * @param timeout Timeout for one invocation of the request executed to a remote instance.
-   */
-  public GrpcRequestForwarder(
-      Consumer consumer,
-      GrantManager<R> grantManager,
-      GrpcStubsCache<STUB> stubsCache,
-      RemoteHandler<REQ, RESP, STUB> remoteHandler,
-      OwnershipRetryer retryer,
-      Duration timeout) {
-    this(consumer, grantManager, remoteHandler, retryer, timeout, stubsCache::getStub);
-  }
-
   /**
    * Creates a new forwarder.
    *
@@ -86,27 +62,11 @@ public GrpcRequestForwarder(
       RemoteHandler<REQ, RESP, STUB> remoteHandler,
       OwnershipRetryer retryer,
       Duration timeout) {
-    this(
-        consumer,
-        grantManager,
-        remoteHandler,
-        retryer,
-        timeout,
-        instance -> stubFactory.apply(consumer.getPeerChannels().channel(instance)));
-  }
-
-  private GrpcRequestForwarder(
-      Consumer consumer,
-      GrantManager<R> grantManager,
-      RemoteHandler<REQ, RESP, STUB> remoteHandler,
-      OwnershipRetryer retryer,
-      Duration timeout,
-      Function<Instance, STUB> stubProvider) {
     this.consumer = consumer;
     this.grantManager = grantManager;
     this.remoteHandler = remoteHandler;
     this.retryer = retryer;
-    this.stubProvider = stubProvider;
+    this.stubProvider = instance -> stubFactory.apply(consumer.getPeerChannels().channel(instance));
     this.timeout = timeout;
   }
 

src/consumer/processor/GrpcRequestForwarderTest.java

@@ -29,14 +29,11 @@
 import co.atoms.splitter.testing.TestStreamObserver;
 import io.grpc.CallOptions;
 import io.grpc.Channel;
-import io.grpc.ManagedChannel;
-import io.grpc.ManagedChannelBuilder;
 import io.grpc.stub.AbstractBlockingStub;
 import io.grpc.stub.StreamObserver;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import org.junit.jupiter.api.BeforeEach;
@@ -233,7 +230,7 @@ public void handleWithActiveLocal(TestInfo testInfo) {
   }
 
   @Test
-  public void handleRemoteWithExistingStubCache(TestInfo testInfo) {
+  public void handleRemoteWithSharedPeerChannels(TestInfo testInfo) {
     var processor = new TestWorkProcessor((id, shard, ownership) -> new TestRange());
     var consumer =
         TestConsumer.create(
@@ -244,23 +241,18 @@ public void handleRemoteWithExistingStubCache(TestInfo testInfo) {
     var grant = GrantInfo.create("g1", SHARD_D2_GLOBAL_5c, ALLOCATED_LOADED);
     consumer.sendClusterSnapshot(MoreMessages.newAssignment(instance, grant));
 
-    var dialed = new ArrayList<String>();
-    var channel = ManagedChannelBuilder.forTarget(instance.endpoint()).usePlaintext().build();
-    var cache =
-        new GrpcStubsCache<>(
-            managedChannel -> new TestStub(managedChannel, CallOptions.DEFAULT),
-            target -> {
-              dialed.add(target);
-              return channel;
-            });
+    var channel = consumer.getPeerChannels().channel(instance);
+    var stubs = new ArrayList<TestStub>();
     try {
-      var stub = cache.getStub(instance);
-      assertSame(stub, cache.getStub(instance));
       var forwarder =
           new GrpcRequestForwarder<>(
               consumer,
               processor,
-              cache,
+              managedChannel -> {
+                var stub = new TestStub(managedChannel, CallOptions.DEFAULT);
+                stubs.add(stub);
+                return stub;
+              },
               (remote, request) -> {
                 assertSame(channel, remote.getChannel());
                 var deadline = remote.getCallOptions().getDeadline();
@@ -273,32 +265,21 @@ public void handleRemoteWithExistingStubCache(TestInfo testInfo) {
       var key = Fixtures.newKey(DOMAIN2, "5");
       assertForwarded(o -> forwarder.forward(key, key, o), key);
       assertForwarded(o -> forwarder.forward(key, key, o), key);
-      assertEquals(List.of(instance.endpoint()), dialed);
-      assertNull(stub.getCallOptions().getDeadline());
+      assertEquals(2, stubs.size());
+      for (var stub : stubs) {
+        assertSame(channel, stub.getChannel());
+        assertNull(stub.getCallOptions().getDeadline());
+      }
 
       consumer.sendClusterRemove(1, instance.id());
-      assertFalse(channel.isShutdown());
+      assertCondition(channel::isShutdown);
       consumer.stop();
       consumer.awaitTerminated();
-      // Existing caches keep ownership of their channels, even after consumer shutdown.
-      assertFalse(channel.isShutdown());
     } finally {
       channel.shutdownNow();
     }
   }
 
-  @Test
-  public void existingStubCacheDefaultConstructor() {
-    var cache = new GrpcStubsCache<>(channel -> new TestStub(channel, CallOptions.DEFAULT));
-    var instance = Fixtures.newInstance("remote");
-    var stub = cache.getStub(instance);
-    try {
-      assertSame(stub, cache.getStub(instance));
-    } finally {
-      ((ManagedChannel) stub.getChannel()).shutdownNow();
-    }
-  }
-
   @Test
   public void handleRetry(TestInfo testInfo) {
     var r = new TestRange();