diff --git a/src/consumer/Consumer.java b/src/consumer/Consumer.java index 1fc77dc..c1b3a99 100644 --- a/src/consumer/Consumer.java +++ b/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); diff --git a/src/consumer/processor/GrpcRequestForwarder.java b/src/consumer/processor/GrpcRequestForwarder.java index 8e8e970..f9623b1 100644 --- a/src/consumer/processor/GrpcRequestForwarder.java +++ b/src/consumer/processor/GrpcRequestForwarder.java @@ -43,30 +43,6 @@ public interface RemoteHandlerThe 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 grantManager, - GrpcStubsCache stubsCache, - RemoteHandler remoteHandler, - OwnershipRetryer retryer, - Duration timeout) { - this(consumer, grantManager, remoteHandler, retryer, timeout, stubsCache::getStub); - } - /** * Creates a new forwarder. * @@ -86,27 +62,11 @@ public GrpcRequestForwarder( RemoteHandler remoteHandler, OwnershipRetryer retryer, Duration timeout) { - this( - consumer, - grantManager, - remoteHandler, - retryer, - timeout, - instance -> stubFactory.apply(consumer.getPeerChannels().channel(instance))); - } - - private GrpcRequestForwarder( - Consumer consumer, - GrantManager grantManager, - RemoteHandler remoteHandler, - OwnershipRetryer retryer, - Duration timeout, - Function 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; } diff --git a/src/consumer/processor/GrpcRequestForwarderTest.java b/src/consumer/processor/GrpcRequestForwarderTest.java index 1c0cc38..230af92 100644 --- a/src/consumer/processor/GrpcRequestForwarderTest.java +++ b/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(); - 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(); 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(); diff --git a/src/consumer/processor/GrpcStubsCache.java b/src/consumer/processor/GrpcStubsCache.java deleted file mode 100644 index 1abb241..0000000 --- a/src/consumer/processor/GrpcStubsCache.java +++ /dev/null @@ -1,72 +0,0 @@ -package co.atoms.splitter.consumer.processor; - -import co.atoms.splitter.model.Instance; -import com.github.benmanes.caffeine.cache.Caffeine; -import com.github.benmanes.caffeine.cache.LoadingCache; -import io.grpc.ManagedChannel; -import java.time.Duration; -import java.util.function.Function; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; - -/** - * Cache for gRPC stubs to avoid creating new channels on each request. Entries expire ten minutes - * after last access; this cache does not track consumer membership. - * - *

Retained for compatibility. New callers should use the consumer's {@link - * co.atoms.splitter.consumer.PeerChannels} and create stubs over its shared channels. - */ -public class GrpcStubsCache { - - private static final Logger LOGGER = LoggerFactory.getLogger(GrpcStubsCache.class); - - private final LoadingCache> stubs; - private final Function stubFactory; - private final Function channelFactory; - - public GrpcStubsCache(Function stubFactory, Function channelFactory) { - this.stubFactory = stubFactory; - this.channelFactory = channelFactory; - stubs = - Caffeine.newBuilder() - .recordStats() - .maximumSize(100) - .expireAfterAccess(Duration.ofMinutes(10)) - .>removalListener( - (k, value, c) -> this.onCacheRemoval(value)) - .build(this::buildStub); - } - - public GrpcStubsCache(Function stubFactory) { - this(stubFactory, t -> NettyChannelBuilder.forTarget(t).build()); - } - - public STUB getStub(Instance consumer) { - return stubs.get(consumer).stub; - } - - private StubWithChannel buildStub(Instance instance) { - var channel = channelFactory.apply(instance.endpoint()); - return new StubWithChannel<>(channel, stubFactory.apply(channel)); - } - - private void onCacheRemoval(StubWithChannel stub) { - try { - stub.channel.shutdown(); - } catch (Exception error) { - LOGGER.error("Failed to shut down channel", error); - } - } - - private static class StubWithChannel { - final ManagedChannel channel; - final STUB stub; - - public StubWithChannel(ManagedChannel channel, STUB stub) { - this.channel = channel; - this.stub = stub; - } - } -}