diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcConfigKeys.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcConfigKeys.java index 451ba19231..9b13e13baf 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcConfigKeys.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcConfigKeys.java @@ -332,18 +332,6 @@ static void setCredentials(Parameters parameters, ServerCredentials credentials) parameters.put(CREDENTIALS_PARAMETER, credentials, CREDENTIALS_CLASS); } - String LOG_APPENDER_LISTENER_FACTORY_PARAMETER = PREFIX + ".log.appender.listener.factory"; - Class LOG_APPENDER_LISTENER_FACTORY_CLASS = GrpcLogAppenderListener.Factory.class; - static GrpcLogAppenderListener.Factory logAppenderListenerFactory(Parameters parameters) { - return parameters == null ? null : parameters.get( - LOG_APPENDER_LISTENER_FACTORY_PARAMETER, LOG_APPENDER_LISTENER_FACTORY_CLASS); - } - - /** Sets an optional factory for observing the lifecycle of each peer log appender. */ - static void setLogAppenderListenerFactory(Parameters parameters, GrpcLogAppenderListener.Factory factory) { - parameters.put(LOG_APPENDER_LISTENER_FACTORY_PARAMETER, factory, LOG_APPENDER_LISTENER_FACTORY_CLASS); - } - String TLS_CONF_PARAMETER = PREFIX + ".tls.conf"; Class TLS_CONF_CLASS = TLS.CONF_CLASS; static GrpcTlsConfig tlsConf(Parameters parameters) { @@ -362,12 +350,27 @@ static void setStubPoolSize(RaftProperties properties, int size) { setInt(properties::setInt, STUB_POOL_SIZE_KEY, size); } + String LOG_APPENDER_LISTENER_FACTORY_PARAMETER = PREFIX + ".log.appender.listener.factory"; + Class LOG_APPENDER_LISTENER_FACTORY_CLASS = GrpcLogAppenderListener.Factory.class; + static GrpcLogAppenderListener.Factory logAppenderListenerFactory(Parameters parameters) { + return parameters == null ? null : parameters.get( + LOG_APPENDER_LISTENER_FACTORY_PARAMETER, LOG_APPENDER_LISTENER_FACTORY_CLASS); + } + /** Sets an optional factory for observing the lifecycle of each peer log appender. */ + static void setLogAppenderListenerFactory(Parameters parameters, GrpcLogAppenderListener.Factory factory) { + parameters.put(LOG_APPENDER_LISTENER_FACTORY_PARAMETER, factory, LOG_APPENDER_LISTENER_FACTORY_CLASS); + } + /** - * The number of connections to each peer used for appendEntries. The groups shared with a peer are spread over - * them, each group always using the same one, so that the replication between two servers is not limited to one - * connection and so one event-loop thread at each end. Heartbeats keep their own channel. + * The number of connections to each peer for appendEntries. + * When it is 1 (the default), appendEntries uses the main connection to the peer, + * which is shared with the other RPCs such as requestVote and installSnapshot. + * Otherwise, appendEntries uses this many extra connections dedicated to it, but not the main connection. + * The connections are shared by the appendEntries streams of all the groups with the peer; + * each stream uses one connection and the new streams take the connections in turn. + * When there is a separate heartbeat channel (see {@link #HEARTBEAT_CHANNEL_KEY}), heartbeats do not use them. */ - String LOG_APPENDER_CONNECTIONS_KEY = PREFIX + ".log-appender.connections"; + String LOG_APPENDER_CONNECTIONS_KEY = PREFIX + ".log.appender.connections"; int LOG_APPENDER_CONNECTIONS_DEFAULT = 1; static int logAppenderConnections(RaftProperties properties) { return getInt(properties::getInt, LOG_APPENDER_CONNECTIONS_KEY, diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcAppendEntriesChannels.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcAppendEntriesChannels.java new file mode 100644 index 0000000000..ca825a1edb --- /dev/null +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcAppendEntriesChannels.java @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.grpc.server; + +import org.apache.ratis.proto.RaftProtos.AppendEntriesReplyProto; +import org.apache.ratis.proto.RaftProtos.AppendEntriesRequestProto; +import org.apache.ratis.proto.grpc.RaftServerProtocolServiceGrpc; +import org.apache.ratis.proto.grpc.RaftServerProtocolServiceGrpc.RaftServerProtocolServiceStub; +import org.apache.ratis.thirdparty.io.grpc.ManagedChannel; +import org.apache.ratis.thirdparty.io.grpc.stub.CallStreamObserver; +import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; +import org.apache.ratis.util.Preconditions; + +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; + +/** + * Channels for creating appendEntries streams to a peer. + *

+ * When there are no other channels, use only the main channel to create streams, + * which is shared with the other RPCs (requestVote, installSnapshot, etc.). + * Otherwise, use only the other channels, which are dedicated to appendEntries; + * each stream uses one channel and the new streams take the channels in turn. + *

+ * All the channels, including the other channels, are owned by the caller, + * which is also responsible for shutting them down. + */ +class GrpcAppendEntriesChannels { + // The main channel; null if the other channels are used for appendEntries. + private final ManagedChannel mainChannel; + private final RaftServerProtocolServiceStub mainStub; + // The channels dedicated to appendEntries; null if the main channel is used for appendEntries. + private final List otherChannels; + private final List otherStubs; + // The index of the channel in otherChannels for the next stream + private final AtomicInteger channelIndex = new AtomicInteger(); + + /** + * @param mainChannel the main channel, used for appendEntries only if there are no other channels; + * otherwise, it is ignored. + * @param otherChannels the channels dedicated to appendEntries; + * null means that there are no other channels, otherwise it must be non-empty. + */ + GrpcAppendEntriesChannels(ManagedChannel mainChannel, List otherChannels) { + if (otherChannels == null) { + this.mainChannel = mainChannel; + this.mainStub = RaftServerProtocolServiceGrpc.newStub(mainChannel); + this.otherChannels = null; + this.otherStubs = null; + } else { + this.mainChannel = null; + this.mainStub = null; + Preconditions.assertTrue(!otherChannels.isEmpty(), "otherChannels is empty"); + this.otherChannels = Collections.unmodifiableList(otherChannels); + this.otherStubs = Collections.unmodifiableList(otherChannels.stream() + .map(RaftServerProtocolServiceGrpc::newStub) + .collect(Collectors.toList())); + } + } + + /** Create a new {@link CallStreamObserver} for appendEntries. */ + CallStreamObserver newStream(StreamObserver responseHandler) { + final RaftServerProtocolServiceStub stub = otherStubs != null ? otherStubs.get(nextChannelIndex()) : mainStub; + return (CallStreamObserver) stub.appendEntries(responseHandler); + } + + /** + * This method must be called only when there are other channels. + * + * @return the index in the other channels for a new stream. + */ + int nextChannelIndex() { + Preconditions.assertNotNull(otherStubs, "The method should not be called when otherStubs == null"); + return channelIndex.getAndUpdate(n -> (n + 1) % otherStubs.size()); + } + + List getAppendEntriesChannelsForTesting() { + return otherChannels != null ? otherChannels : Collections.singletonList(mainChannel); + } + + List getOtherChannelsForTesting() { + return otherChannels; + } +} diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java index d1cdb2bb7d..69ae1de525 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java @@ -34,7 +34,6 @@ import org.apache.ratis.server.protocol.TermIndex; import org.apache.ratis.server.raftlog.RaftLog; import org.apache.ratis.server.util.ServerStringUtils; -import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting; import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException; import org.apache.ratis.thirdparty.io.grpc.stub.CallStreamObserver; import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; @@ -164,8 +163,6 @@ synchronized int process(Event event) { private final TimeoutExecutor scheduler = TimeoutExecutor.getInstance(); @SuppressWarnings({"squid:S3077"}) // Suppress volatile for generic type private volatile StreamObservers appendLogRequestObserver; - // the appendEntries channel of this appender, taken when it creates its first stream; guarded by the write lock - private volatile int appendEntriesIndex = -1; private final boolean useSeparateHBChannel; private final GrpcServerMetrics grpcServerMetrics; @@ -392,10 +389,10 @@ static class StreamObservers { private final TimeDuration waitForReady; private volatile boolean running = true; - StreamObservers(GrpcServerProtocolClient client, int appendEntriesIndex, AppendLogResponseHandler handler, + StreamObservers(GrpcServerProtocolClient client, AppendLogResponseHandler handler, boolean separateHeartbeat, TimeDuration waitTimeMin) { - this.appendLog = client.appendEntries(appendEntriesIndex, handler, false); - this.heartbeat = separateHeartbeat? client.appendEntries(appendEntriesIndex, handler, true): null; + this.appendLog = client.appendEntries(handler, false); + this.heartbeat = separateHeartbeat? client.appendEntries(handler, true): null; this.waitForReady = waitTimeMin.isPositive()? waitTimeMin: TimeDuration.ONE_MILLISECOND; } @@ -425,11 +422,6 @@ void onCompleted() { } } - @VisibleForTesting - int getAppendEntriesIndex() { - return appendEntriesIndex; - } - @Override public long getCallId() { return callId.get(); @@ -456,11 +448,7 @@ void appendLog(boolean heartbeat) throws IOException { notifyAppendEntriesListener(l -> l.onRequest(pending)); increaseNextIndex(pending); if (appendLogRequestObserver == null) { - final GrpcServerProtocolClient client = getClient(); - if (appendEntriesIndex < 0) { - appendEntriesIndex = client.nextAppendEntriesIndex(); - } - appendLogRequestObserver = new StreamObservers(client, appendEntriesIndex, + appendLogRequestObserver = new StreamObservers(getClient(), new AppendLogResponseHandler(), useSeparateHBChannel, getWaitTimeMin()); } } diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java index 915cb30316..b8eaead8b3 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java @@ -22,7 +22,6 @@ import org.apache.ratis.grpc.util.StreamObserverWithTimeout; import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.server.util.ServerStringUtils; -import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting; import org.apache.ratis.thirdparty.io.grpc.ManagedChannel; import org.apache.ratis.thirdparty.io.grpc.netty.NegotiationType; import org.apache.ratis.thirdparty.io.grpc.netty.NettyChannelBuilder; @@ -43,7 +42,6 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; /** * This is a RaftClient implementation that supports streaming data to the raft @@ -52,18 +50,14 @@ class GrpcServerProtocolClient implements Closeable { // Common channel private final ManagedChannel channel; - private final GrpcStubPool pool; - // Channel and stub for heartbeat - private ManagedChannel hbChannel; - private RaftServerProtocolServiceStub hbAsyncStub; private final RaftServerProtocolServiceStub asyncStub; - // Channels and stubs for appendEntries, each group always using the same one; the first is the main channel - private final List appendEntriesChannels; - private final List appendEntriesStubs; - // The appendEntries channel for the next new log appender, so that the appenders take the channels in turn - private final AtomicInteger nextAppendEntriesIndex = new AtomicInteger(); + private final RaftServerProtocolServiceStub hbAsyncStub; + // All the channels: the main channel, the other appendEntries channels and the heartbeat channel + private final List allChannels; + // The appendEntries streams, using either the main channel or the other channels dedicated to appendEntries + private final GrpcAppendEntriesChannels appendEntriesStreams; + private final GrpcStubPool pool; private final RaftServerProtocolServiceBlockingStub blockingStub; - private final boolean useSeparateHBChannel; private final TimeDuration requestTimeoutDuration; private static final Logger LOG = LoggerFactory.getLogger(GrpcServerProtocolClient.class); @@ -77,28 +71,34 @@ class GrpcServerProtocolClient implements Closeable { EventLoopGroup eventLoopGroup) { raftPeerId = target.getId(); LOG.info("Build channel for {}", target); - useSeparateHBChannel = separateHBChannel; this.eventLoopGroup = eventLoopGroup; channel = buildChannel(target, flowControlWindow, sslContext); blockingStub = RaftServerProtocolServiceGrpc.newBlockingStub(channel); asyncStub = RaftServerProtocolServiceGrpc.newStub(channel); - final List channels = new ArrayList<>(appendEntriesConnections); - final List stubs = new ArrayList<>(appendEntriesConnections); - channels.add(channel); - stubs.add(asyncStub); - for (int i = 1; i < appendEntriesConnections; i++) { - final ManagedChannel c = buildChannel(target, flowControlWindow, sslContext); - channels.add(c); - stubs.add(RaftServerProtocolServiceGrpc.newStub(c)); + + final List all = new ArrayList<>(); + all.add(channel); + List otherChannels = null; + if (appendEntriesConnections > 1) { + otherChannels = new ArrayList<>(); + for (int i = 0; i < appendEntriesConnections; i++) { + otherChannels.add(buildChannel(target, flowControlWindow, sslContext)); + } + all.addAll(otherChannels); } - appendEntriesChannels = Collections.unmodifiableList(channels); - appendEntriesStubs = Collections.unmodifiableList(stubs); - if (useSeparateHBChannel) { - hbChannel = buildChannel(target, flowControlWindow, sslContext); + this.appendEntriesStreams = new GrpcAppendEntriesChannels(channel, otherChannels); + + // Channel and stub for heartbeat + if (separateHBChannel) { + final ManagedChannel hbChannel = buildChannel(target, flowControlWindow, sslContext); + all.add(hbChannel); hbAsyncStub = RaftServerProtocolServiceGrpc.newStub(hbChannel); + } else { + hbAsyncStub = null; } requestTimeoutDuration = requestTimeout; this.pool = connections == 1? null : newGrpcStubPool(target.getAddress(), sslContext, connections); + this.allChannels = Collections.unmodifiableList(all); } GrpcStubPool newGrpcStubPool(String address, SslContext sslContext, int connections) { @@ -128,12 +128,7 @@ private ManagedChannel buildChannel(RaftPeer target, int flowControlWindow, SslC @Override public void close() { LOG.info("{} Close channels", raftPeerId); - // the first one is the main channel - final List channels = new ArrayList<>(appendEntriesChannels); - if (useSeparateHBChannel) { - channels.add(hbChannel); - } - GrpcUtil.shutdownManagedChannels(channels); + GrpcUtil.shutdownManagedChannels(allChannels); if (pool != null) { pool.close(); } @@ -195,32 +190,25 @@ public void onCompleted() { } } - /** @param appendEntriesIndex the appendEntries channel, from {@link #nextAppendEntriesIndex()} */ - CallStreamObserver appendEntries(int appendEntriesIndex, + /** + * Create a new appendEntries stream. + * A heartbeat stream uses the heartbeat channel if there is a separate one. + * Otherwise, the stream is created by {@link GrpcAppendEntriesChannels#newStream(StreamObserver)}. + */ + CallStreamObserver appendEntries( StreamObserver responseHandler, boolean isHeartbeat) { - if (isHeartbeat && useSeparateHBChannel) { + if (isHeartbeat && hbAsyncStub != null) { return (CallStreamObserver) hbAsyncStub.appendEntries(responseHandler); } else { - final RaftServerProtocolServiceStub stub = appendEntriesStubs.get(appendEntriesIndex); - return (CallStreamObserver) stub.appendEntries(responseHandler); + return appendEntriesStreams.newStream(responseHandler); } } - /** - * @return the index of the appendEntries channel for a new log appender. The appenders take the channels in turn; - * an appender keeps its channel, so the appendEntries of a group to this peer use one channel. - */ - int nextAppendEntriesIndex() { - return Math.floorMod(nextAppendEntriesIndex.getAndIncrement(), appendEntriesStubs.size()); + GrpcAppendEntriesChannels getAppendEntriesStreamsForTesting() { + return appendEntriesStreams; } - @VisibleForTesting - List getAppendEntriesChannels() { - return appendEntriesChannels; - } - - @VisibleForTesting - ManagedChannel getMainChannel() { + ManagedChannel getMainChannelForTesting() { return channel; } @@ -232,10 +220,6 @@ StreamObserver installSnapshot( // short-circuit the backoff timer and make them reconnect immediately. public void resetConnectBackoff() { - if (useSeparateHBChannel) { - hbChannel.resetConnectBackoff(); - } - // the first one is the main channel - appendEntriesChannels.forEach(ManagedChannel::resetConnectBackoff); + allChannels.forEach(ManagedChannel::resetConnectBackoff); } } diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java index 3285893017..79ae4853bf 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderCallbacks.java @@ -62,7 +62,6 @@ import java.util.concurrent.TimeUnit; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.RETURNS_DEEP_STUBS; @@ -189,7 +188,7 @@ public void testStreamCreationFailure(boolean failProxy) throws Exception { when(serverRpc.getProxies().getProxy(destination)).thenThrow(error); } else { error = new IllegalStateException("Failed to create append stream"); - when(client.appendEntries(anyInt(), any(), eq(false))).thenThrow(error); + when(client.appendEntries(any(), eq(false))).thenThrow(error); } Assertions.assertSame(error, Assertions.assertThrows(Exception.class, () -> appender.appendLog(false))); final InOrder order = inOrder(appendEntries); diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java index bf817c37bb..ce5204e3ee 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcLogAppenderConnections.java @@ -18,16 +18,14 @@ package org.apache.ratis.grpc.server; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.ArrayList; -import java.util.HashMap; +import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Set; -import java.util.TreeSet; import java.util.stream.Collectors; -import java.util.stream.IntStream; import org.apache.ratis.BaseTest; import org.apache.ratis.RaftTestUtil; import org.apache.ratis.client.RaftClient; @@ -43,6 +41,8 @@ import org.apache.ratis.server.impl.RaftServerTestUtil; import org.apache.ratis.server.leader.LogAppender; import org.apache.ratis.RaftTestUtil.SimpleMessage; +import org.apache.ratis.thirdparty.io.grpc.ConnectivityState; +import org.apache.ratis.thirdparty.io.grpc.ManagedChannel; import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.TimeDuration; import org.junit.jupiter.api.Test; @@ -100,20 +100,26 @@ private static void assertConnections(MiniRaftCluster cluster, RaftPeerId leader throws Exception { final RaftServer leader = cluster.getServer(leaderId); final GrpcServicesImpl rpc = (GrpcServicesImpl) leader.getServerRpc(); - final Map> indices = new HashMap<>(); + final Set followers = new HashSet<>(); for (RaftGroup group : groups) { final List appenders = RaftServerTestUtil.getLogAppenders( cluster.getDivision(leaderId, group.getGroupId())).collect(Collectors.toList()); assertEquals(group.getPeers().size() - 1, appenders.size()); - for (LogAppender appender : appenders) { - final RaftPeerId follower = appender.getFollowerId(); - assertEquals(CONNECTIONS, rpc.getProxies().getProxy(follower).getAppendEntriesChannels().size()); - final int index = ((GrpcLogAppender) appender).getAppendEntriesIndex(); - indices.computeIfAbsent(follower, k -> new TreeSet<>()).add(index); + appenders.forEach(appender -> followers.add(appender.getFollowerId())); + } + assertEquals(2, followers.size()); + for (RaftPeerId follower : followers) { + final GrpcServerProtocolClient client = rpc.getProxies().getProxy(follower); + final List channels = client.getAppendEntriesStreamsForTesting().getOtherChannelsForTesting(); + assertEquals(CONNECTIONS, channels.size()); + // the main channel is not used for appendEntries + assertFalse(channels.contains(client.getMainChannelForTesting())); + assertEquals(channels, client.getAppendEntriesStreamsForTesting().getAppendEntriesChannelsForTesting()); + // The channels connect lazily, so a channel is READY only if some stream has used it. + for (int i = 0; i < channels.size(); i++) { + assertEquals(ConnectivityState.READY, channels.get(i).getState(false), + leaderId + " -> " + follower + ": channel " + i + " is not used"); } } - final Set all = IntStream.range(0, CONNECTIONS).boxed().collect(Collectors.toCollection(TreeSet::new)); - assertEquals(2, indices.size()); - indices.forEach((follower, used) -> assertEquals(all, used, leaderId + " -> " + follower)); } } diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java index e863ca7792..531d2f6a6b 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerProtocolClientChannels.java @@ -18,9 +18,11 @@ package org.apache.ratis.grpc.server; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.concurrent.TimeUnit; @@ -42,34 +44,46 @@ private static GrpcServerProtocolClient newClient(int appendEntriesConnections) @Test public void testOneConnectionUsesMainChannel() { final GrpcServerProtocolClient client = newClient(1); + final GrpcAppendEntriesChannels streams = client.getAppendEntriesStreamsForTesting(); try { - final List channels = client.getAppendEntriesChannels(); - assertEquals(1, channels.size()); - assertSame(client.getMainChannel(), channels.get(0)); - for (int i = 0; i < 10; i++) { - assertEquals(0, client.nextAppendEntriesIndex()); - } + // no other channels; appendEntries uses the main channel + assertNull(streams.getOtherChannelsForTesting()); + assertEquals(Collections.singletonList(client.getMainChannelForTesting()), + streams.getAppendEntriesChannelsForTesting()); + + // called by GrpcLogAppender when it resets the client + client.resetConnectBackoff(); } finally { client.close(); } + assertTrue(client.getMainChannelForTesting().isShutdown(), "the main channel is not shut down"); } @Test - public void testAppendersTakeConnectionsInTurn() { + public void testStreamsTakeConnectionsInTurn() { final int connections = 4; final GrpcServerProtocolClient client = newClient(connections); - final List channels = client.getAppendEntriesChannels(); + final GrpcAppendEntriesChannels streams = client.getAppendEntriesStreamsForTesting(); + final List others = streams.getOtherChannelsForTesting(); try { - assertEquals(connections, channels.size()); - assertSame(client.getMainChannel(), channels.get(0)); - assertEquals(connections, new HashSet<>(channels).size()); + // the channels dedicated to appendEntries + assertEquals(connections, others.size()); + assertEquals(connections, new HashSet<>(others).size()); + + // the main channel is not used for appendEntries + assertFalse(others.contains(client.getMainChannelForTesting())); + assertEquals(others, streams.getAppendEntriesChannelsForTesting()); for (int i = 0; i < 100; i++) { - assertEquals(i % connections, client.nextAppendEntriesIndex()); + assertEquals(i % connections, streams.nextChannelIndex()); } + + // called by GrpcLogAppender when it resets the client + client.resetConnectBackoff(); } finally { client.close(); } - channels.forEach(c -> assertTrue(c.isShutdown(), c + " is not shut down")); + assertTrue(client.getMainChannelForTesting().isShutdown(), "the main channel is not shut down"); + others.forEach(c -> assertTrue(c.isShutdown(), c + " is not shut down")); } }