Repository: incubator-ratis Updated Branches: refs/heads/master ba83ce132 -> 451e8551b
RATIS-231. Grpc server should reconnect if there is an exception. Project: http://git-wip-us.apache.org/repos/asf/incubator-ratis/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ratis/commit/451e8551 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ratis/tree/451e8551 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ratis/diff/451e8551 Branch: refs/heads/master Commit: 451e8551b7f8aea23a969d6bccb8a56b5937c7f7 Parents: ba83ce1 Author: Tsz Sze <[email protected]> Authored: Mon Apr 30 10:50:03 2018 -0700 Committer: Tsz Sze <[email protected]> Committed: Mon Apr 30 10:50:03 2018 -0700 ---------------------------------------------------------------------- .../apache/ratis/protocol/RaftClientReply.java | 3 +- .../org/apache/ratis/grpc/RaftGRpcService.java | 88 ++------- .../ratis/grpc/server/GRpcLogAppender.java | 187 +++++++++---------- .../grpc/server/RaftServerProtocolClient.java | 7 +- .../ratis/grpc/MiniRaftClusterWithGRpc.java | 11 -- .../ratis/grpc/TestRaftServerWithGrpc.java | 25 +-- .../apache/ratis/server/impl/LogAppender.java | 24 +-- .../ratis/server/impl/RaftServerImpl.java | 9 +- .../ratis/server/impl/RaftServerProxy.java | 9 +- .../apache/ratis/server/impl/ServerState.java | 20 +- .../server/storage/RaftStorageDirectory.java | 2 +- .../java/org/apache/ratis/MiniRaftCluster.java | 10 +- .../impl/RaftReconfigurationBaseTest.java | 8 +- 13 files changed, 172 insertions(+), 231 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientReply.java ---------------------------------------------------------------------- diff --git a/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientReply.java b/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientReply.java index af64d66..8254ab4 100644 --- a/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientReply.java +++ b/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientReply.java @@ -24,6 +24,7 @@ import org.apache.ratis.util.ProtoUtils; import org.apache.ratis.util.ReflectionUtils; import java.util.Collection; +import java.util.Collections; import java.util.stream.Collector; import java.util.stream.Collectors; @@ -63,7 +64,7 @@ public class RaftClientReply extends RaftClientMessage { () -> "Unexpected exception class: " + this); } - this.commitInfos = commitInfos; + this.commitInfos = commitInfos != null? commitInfos: Collections.emptyList(); } public RaftClientReply(RaftClientRequest request, RaftException exception, Collection<CommitInfoProto> commitInfos) { http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-grpc/src/main/java/org/apache/ratis/grpc/RaftGRpcService.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/RaftGRpcService.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/RaftGRpcService.java index d3827ef..fefec48 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/RaftGRpcService.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/RaftGRpcService.java @@ -21,33 +21,26 @@ import org.apache.ratis.grpc.client.RaftClientProtocolService; import org.apache.ratis.grpc.server.AdminProtocolService; import org.apache.ratis.grpc.server.RaftServerProtocolClient; import org.apache.ratis.grpc.server.RaftServerProtocolService; -import org.apache.ratis.protocol.RaftPeer; import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.rpc.SupportedRpcType; import org.apache.ratis.server.RaftServer; import org.apache.ratis.server.RaftServerConfigKeys; import org.apache.ratis.server.RaftServerRpc; +import org.apache.ratis.server.impl.RaftServerRpcWithProxy; import org.apache.ratis.shaded.io.grpc.Server; import org.apache.ratis.shaded.io.grpc.ServerBuilder; import org.apache.ratis.shaded.io.grpc.netty.NettyServerBuilder; import org.apache.ratis.shaded.proto.RaftProtos.*; -import org.apache.ratis.util.CodeInjectionForTesting; -import org.apache.ratis.util.ExitUtils; -import org.apache.ratis.util.SizeInBytes; -import org.apache.ratis.util.TimeDuration; +import org.apache.ratis.util.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.net.InetSocketAddress; -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; -import java.util.Objects; import java.util.function.Supplier; /** A grpc implementation of {@link RaftServerRpc}. */ -public class RaftGRpcService implements RaftServerRpc { +public class RaftGRpcService extends RaftServerRpcWithProxy<RaftServerProtocolClient, PeerProxyMap<RaftServerProtocolClient>> { static final Logger LOG = LoggerFactory.getLogger(RaftGRpcService.class); public static final String GRPC_SEND_SERVER_REQUEST = RaftGRpcService.class.getSimpleName() + ".sendRequest"; @@ -71,50 +64,35 @@ public class RaftGRpcService implements RaftServerRpc { } private final Server server; - private final InetSocketAddress address; - private final Map<RaftPeerId, RaftServerProtocolClient> peers = - Collections.synchronizedMap(new HashMap<>()); - private final Supplier<RaftPeerId> idSupplier; - private final int flowControlWindow; - private final TimeDuration requestTimeoutDuration; + private final Supplier<InetSocketAddress> addressSupplier; private RaftGRpcService(RaftServer server) { - this(server, + this(server, server::getId, GrpcConfigKeys.Server.port(server.getProperties()), - GrpcConfigKeys.messageSizeMax(server.getProperties()).getSizeInt(), GrpcConfigKeys.messageSizeMax(server.getProperties()), RaftServerConfigKeys.Log.Appender.bufferCapacity(server.getProperties()), GrpcConfigKeys.flowControlWindow(server.getProperties()), RaftServerConfigKeys.Rpc.requestTimeout(server.getProperties())); } - private RaftGRpcService(RaftServer raftServer, int port, int maxMessageSize, + private RaftGRpcService(RaftServer raftServer, Supplier<RaftPeerId> idSupplier, int port, SizeInBytes grpcMessageSizeMax, SizeInBytes appenderBufferSize, - SizeInBytes flowControlWindowSize, TimeDuration requestTimeoutDuration) { + SizeInBytes flowControlWindow, TimeDuration requestTimeoutDuration) { + super(idSupplier, id -> new PeerProxyMap<>(id.toString(), + p -> new RaftServerProtocolClient(p, flowControlWindow.getSizeInt(), requestTimeoutDuration))); if (appenderBufferSize.getSize() > grpcMessageSizeMax.getSize()) { throw new IllegalArgumentException("Illegal configuration: " + RaftServerConfigKeys.Log.Appender.BUFFER_CAPACITY_KEY + " = " + appenderBufferSize + " > " + GrpcConfigKeys.MESSAGE_SIZE_MAX_KEY + " = " + grpcMessageSizeMax); } - this.flowControlWindow = flowControlWindowSize.getSizeInt(); - this.requestTimeoutDuration = requestTimeoutDuration; - ServerBuilder serverBuilder = ServerBuilder.forPort(port); - idSupplier = raftServer::getId; - server = ((NettyServerBuilder) serverBuilder).maxMessageSize(maxMessageSize) - .flowControlWindow(flowControlWindow) + server = ((NettyServerBuilder) ServerBuilder.forPort(port)) + .maxMessageSize(grpcMessageSizeMax.getSizeInt()) + .flowControlWindow(flowControlWindow.getSizeInt()) .addService(new RaftServerProtocolService(idSupplier, raftServer)) .addService(new RaftClientProtocolService(idSupplier, raftServer)) .addService(new AdminProtocolService(raftServer)) .build(); - - // start service to determine the port (in case port is configured as 0) - startService(); - address = new InetSocketAddress(server.getPort()); - LOG.info("Server started, listening on " + address.getPort()); - } - - RaftPeerId getId() { - return idSupplier.get(); + addressSupplier = JavaUtils.memoize(() -> new InetSocketAddress(port != 0? port: server.getPort())); } @Override @@ -123,34 +101,26 @@ public class RaftGRpcService implements RaftServerRpc { } @Override - public void start() { - // do nothing - } - - private void startService() { + public void startImpl() { try { server.start(); } catch (IOException e) { ExitUtils.terminate(1, "Failed to start Grpc server", e, LOG); } - Runtime.getRuntime().addShutdownHook(new Thread(() -> { - System.err.println("*** shutting down gRPC server since JVM is shutting down"); - RaftGRpcService.this.close(); - System.err.println("*** server shut down"); - })); + LOG.info("{}: {} started, listening on {}", getId(), getClass().getSimpleName(), getInetSocketAddress()); } @Override - public void close() { + public void closeImpl() { if (server != null) { server.shutdown(); } - shutdownClients(); + super.closeImpl(); } @Override public InetSocketAddress getInetSocketAddress() { - return address; + return addressSupplier.get(); } @Override @@ -173,25 +143,7 @@ public class RaftGRpcService implements RaftServerRpc { CodeInjectionForTesting.execute(GRPC_SEND_SERVER_REQUEST, getId(), null, request); - RaftServerProtocolClient target = Objects.requireNonNull( - peers.get(RaftPeerId.valueOf(request.getServerRequest().getReplyId()))); - return target.requestVote(request); - } - - @Override - public void addPeers(Iterable<RaftPeer> newPeers) { - for (RaftPeer p : newPeers) { - if (!peers.containsKey(p.getId())) { - peers.put(p.getId(), new RaftServerProtocolClient(p, flowControlWindow, requestTimeoutDuration)); - } - } - } - - private void shutdownClients() { - peers.values().forEach(RaftServerProtocolClient::shutdown); - } - - public RaftServerProtocolClient getRpcClient(RaftPeer peer) { - return peers.get(peer.getId()); + final RaftPeerId target = RaftPeerId.valueOf(request.getServerRequest().getReplyId()); + return getProxies().getProxy(target).requestVote(request); } } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GRpcLogAppender.java ---------------------------------------------------------------------- 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 d69a897..1c1a754 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 @@ -21,28 +21,20 @@ import org.apache.ratis.grpc.GrpcConfigKeys; import org.apache.ratis.grpc.RaftGRpcService; import org.apache.ratis.grpc.RaftGrpcUtil; import org.apache.ratis.server.RaftServerConfigKeys; -import org.apache.ratis.util.TimeoutScheduler; import org.apache.ratis.server.impl.FollowerInfo; import org.apache.ratis.server.impl.LeaderState; import org.apache.ratis.server.impl.LogAppender; import org.apache.ratis.server.impl.RaftServerImpl; -import org.apache.ratis.server.storage.RaftLogIOException; -import org.apache.ratis.shaded.io.grpc.Status; import org.apache.ratis.shaded.io.grpc.stub.StreamObserver; import org.apache.ratis.shaded.proto.RaftProtos.AppendEntriesReplyProto; import org.apache.ratis.shaded.proto.RaftProtos.AppendEntriesRequestProto; import org.apache.ratis.shaded.proto.RaftProtos.InstallSnapshotReplyProto; import org.apache.ratis.shaded.proto.RaftProtos.InstallSnapshotRequestProto; import org.apache.ratis.statemachine.SnapshotInfo; -import org.apache.ratis.util.CodeInjectionForTesting; -import org.apache.ratis.util.Preconditions; -import org.apache.ratis.util.TimeDuration; - -import java.util.LinkedList; -import java.util.Map; -import java.util.Objects; -import java.util.Queue; -import java.util.UUID; +import org.apache.ratis.util.*; + +import java.io.IOException; +import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; @@ -50,107 +42,126 @@ import java.util.concurrent.atomic.AtomicBoolean; * A new log appender implementation using grpc bi-directional stream API. */ public class GRpcLogAppender extends LogAppender { - private final RaftServerProtocolClient client; + private final RaftGRpcService rpcService; private final Map<Long, AppendEntriesRequestProto> pendingRequests; private final int maxPendingRequestsNum; private long callId = 0; private volatile boolean firstResponseReceived = false; - private final AppendLogResponseHandler appendResponseHandler; - private final InstallSnapshotResponseHandler snapshotResponseHandler; private static TimeDuration requestTimeoutDuration; private volatile StreamObserver<AppendEntriesRequestProto> appendLogRequestObserver; - private StreamObserver<InstallSnapshotRequestProto> snapshotRequestObserver; public GRpcLogAppender(RaftServerImpl server, LeaderState leaderState, FollowerInfo f) { super(server, leaderState, f); - RaftGRpcService rpcService = (RaftGRpcService) server.getServerRpc(); - client = rpcService.getRpcClient(f.getPeer()); + this.rpcService = (RaftGRpcService) server.getServerRpc(); + maxPendingRequestsNum = GrpcConfigKeys.Server.leaderOutstandingAppendsMax( server.getProxy().getProperties()); requestTimeoutDuration = RaftServerConfigKeys.Rpc.requestTimeout(server.getProxy().getProperties()); pendingRequests = new ConcurrentHashMap<>(); + } + + private RaftServerProtocolClient getClient() throws IOException { + return rpcService.getProxies().getProxy(follower.getPeer().getId()); + } - appendResponseHandler = new AppendLogResponseHandler(); - snapshotResponseHandler = new InstallSnapshotResponseHandler(); + private synchronized void resetClient(AppendEntriesRequestProto request) { + rpcService.getProxies().resetProxy(follower.getPeer().getId()); + appendLogRequestObserver = null; + firstResponseReceived = false; + + // clear the pending requests queue and reset the next index of follower + final long nextIndex = request != null && request.hasPreviousLog()? + request.getPreviousLog().getIndex() + 1: raftLog.getStartIndex(); + clearPendingRequests(nextIndex); } @Override public void run() { - while (isAppenderRunning()) { + for(; isAppenderRunning(); mayWait()) { if (shouldSendRequest()) { SnapshotInfo snapshot = shouldInstallSnapshot(); if (snapshot != null) { - installSnapshot(snapshot, snapshotResponseHandler); - } else { + installSnapshot(snapshot); + } else if (!shouldWait()) { // keep appending log entries or sending heartbeats try { appendLog(); - } catch (RaftLogIOException e) { + } catch (IOException e) { LOG.error(this + " hit IOException while loading raft log", e); stopSender(); + return; } } } + } - if (isAppenderRunning() && !shouldSendRequest()) { - // use lastSend time instead of lastResponse time - final long waitTime = getHeartbeatRemainingTime( - follower.getLastRpcTime()); - if (waitTime > 0) { - synchronized (this) { - try { - LOG.debug("{} decides to wait {}ms before appending to {}", - server.getId(), waitTime, follower.getPeer()); - wait(waitTime); - } catch (InterruptedException ignored) { - } - } - } + Optional.ofNullable(appendLogRequestObserver).ifPresent(StreamObserver::onCompleted); + } + + private long getWaitTimeMs() { + if (!shouldSendRequest()) { + return getHeartbeatRemainingTime(); // No requests, wait until heartbeat + } else if (shouldWait()) { + return halfMinTimeoutMs; // Should wait for a short time + } + return 0L; + } + + private void mayWait() { + // use lastSend time instead of lastResponse time + final long waitTimeMs = getWaitTimeMs(); + if (waitTimeMs <= 0L) { + return; + } + + synchronized(this) { + try { + LOG.trace("{}: wait {}ms", this, waitTimeMs); + wait(waitTimeMs); + } catch(InterruptedException ie) { + LOG.warn("Wait interrupted", ie); } } - appendLogRequestObserver.onCompleted(); } - private boolean shouldWait() { - return pendingRequests.size() >= maxPendingRequestsNum || - shouldWaitForFirstResponse(); + @Override + protected boolean shouldSendRequest() { + return appendLogRequestObserver == null || super.shouldSendRequest(); } - private void appendLog() throws RaftLogIOException { - if (appendLogRequestObserver == null) { - appendLogRequestObserver = client.appendEntries(appendResponseHandler); + /** @return true iff not received first response or queue is full. */ + private boolean shouldWait() { + final int size = pendingRequests.size(); + if (size == 0) { + return false; } - AppendEntriesRequestProto pending = null; + return !firstResponseReceived || size >= maxPendingRequestsNum; + } + + private void appendLog() throws IOException { + final AppendEntriesRequestProto pending; final StreamObserver<AppendEntriesRequestProto> s; synchronized (this) { - // if the queue's size >= maxSize, wait - while (isAppenderRunning() && shouldWait()) { - try { - LOG.debug("{} wait to send the next AppendEntries to {}", - server.getId(), follower.getPeer()); - this.wait(); - } catch (InterruptedException ignored) { - } + // prepare and enqueue the append request. note changes on follower's + // nextIndex and ops on pendingRequests should always be associated + // together and protected by the lock + pending = createRequest(callId++); + if (pending == null) { + return; } - - if (isAppenderRunning()) { - // prepare and enqueue the append request. note changes on follower's - // nextIndex and ops on pendingRequests should always be associated - // together and protected by the lock - pending = createRequest(callId++); - if (pending != null) { - pendingRequests.put(pending.getServerRequest().getCallId(), pending); - updateNextIndex(pending); - } + pendingRequests.put(pending.getServerRequest().getCallId(), pending); + updateNextIndex(pending); + if (appendLogRequestObserver == null) { + appendLogRequestObserver = getClient().appendEntries(new AppendLogResponseHandler()); } s = appendLogRequestObserver; } - if (pending != null && isAppenderRunning()) { + if (isAppenderRunning()) { sendRequest(pending, s); } } @@ -169,7 +180,8 @@ public class GRpcLogAppender extends LogAppender { private void timeoutAppendRequest(AppendEntriesRequestProto request) { AppendEntriesRequestProto pendingRequest = pendingRequests.remove(request.getServerRequest().getCallId()); if (pendingRequest != null) { - LOG.info("Timeout executed for append entry request: " + pendingRequest); + final String err = this + ": appendEntries Timeout, request=" + ProtoUtils.toString(pendingRequest.getServerRequest()); + LOG.warn(err); } } @@ -181,14 +193,6 @@ public class GRpcLogAppender extends LogAppender { } /** - * if this is the first append, wait for the response of the first append so - * that we can get the correct next index. - */ - private boolean shouldWaitForFirstResponse() { - return pendingRequests.size() > 0 && !firstResponseReceived; - } - - /** * StreamObserver for handling responses from the follower */ private class AppendLogResponseHandler @@ -241,26 +245,7 @@ public class GRpcLogAppender extends LogAppender { RaftGrpcUtil.warn(LOG, () -> server.getId() + ": Failed appendEntries to " + follower.getPeer().getId(), t); long callId = RaftGrpcUtil.getCallId(t); - synchronized (this) { - final Status cause = Status.fromThrowable(t); - if (cause != null && cause.getCode() == Status.Code.INTERNAL) { - // TODO check other Status. Add sleep to avoid tight loop - LOG.debug("{} restarts Append call to {} due to error {}", - server.getId(), follower.getPeer(), t); - // recreate the StreamObserver - appendLogRequestObserver = client.appendEntries(appendResponseHandler); - // reset firstResponseReceived to false - firstResponseReceived = false; - } - - // clear the pending requests queue and reset the next index of follower - AppendEntriesRequestProto request = pendingRequests.get(callId); - if (request != null) { - final long nextIndex = request.hasPreviousLog() ? - request.getPreviousLog().getIndex() + 1 : raftLog.getStartIndex(); - clearPendingRequests(nextIndex); - } - } + resetClient(pendingRequests.get(callId)); } @Override @@ -395,6 +380,7 @@ public class GRpcLogAppender extends LogAppender { } LOG.info("{} got error when installing snapshot to {}, exception: {}", server.getId(), follower.getPeer(), t); + resetClient(null); close(); } @@ -406,16 +392,17 @@ public class GRpcLogAppender extends LogAppender { } } - private void installSnapshot(SnapshotInfo snapshot, - InstallSnapshotResponseHandler responseHandler) { + private void installSnapshot(SnapshotInfo snapshot) { LOG.info("{}: follower {}'s next index is {}," + " log's start index is {}, need to install snapshot", server.getId(), follower.getPeer(), follower.getNextIndex(), raftLog.getStartIndex()); - snapshotRequestObserver = client.installSnapshot(snapshotResponseHandler); + final InstallSnapshotResponseHandler responseHandler = new InstallSnapshotResponseHandler(); + StreamObserver<InstallSnapshotRequestProto> snapshotRequestObserver = null; final String requestId = UUID.randomUUID().toString(); try { + snapshotRequestObserver = getClient().installSnapshot(responseHandler); for (InstallSnapshotRequestProto request : new SnapshotRequestIter(snapshot, requestId)) { if (isAppenderRunning()) { @@ -430,10 +417,10 @@ public class GRpcLogAppender extends LogAppender { } catch (Exception e) { LOG.warn("{} failed to install snapshot {}. Exception: {}", this, snapshot.getFiles(), e); - snapshotRequestObserver.onError(e); + if (snapshotRequestObserver != null) { + snapshotRequestObserver.onError(e); + } return; - } finally { - snapshotRequestObserver = null; } synchronized (this) { http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java index 034f06c..b801c2a 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java @@ -27,11 +27,13 @@ import org.apache.ratis.shaded.proto.grpc.RaftServerProtocolServiceGrpc.RaftServ import org.apache.ratis.protocol.RaftPeer; import org.apache.ratis.util.TimeDuration; +import java.io.Closeable; + /** * This is a RaftClient implementation that supports streaming data to the raft * ring. The stream implementation utilizes gRPC. */ -public class RaftServerProtocolClient { +public class RaftServerProtocolClient implements Closeable { private final ManagedChannel channel; private final TimeDuration requestTimeoutDuration; private final RaftServerProtocolServiceBlockingStub blockingStub; @@ -47,7 +49,8 @@ public class RaftServerProtocolClient { this.requestTimeoutDuration = requestTimeoutDuration; } - public void shutdown() { + @Override + public void close() { channel.shutdownNow(); } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-grpc/src/test/java/org/apache/ratis/grpc/MiniRaftClusterWithGRpc.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/test/java/org/apache/ratis/grpc/MiniRaftClusterWithGRpc.java b/ratis-grpc/src/test/java/org/apache/ratis/grpc/MiniRaftClusterWithGRpc.java index 3580f22..a2c419f 100644 --- a/ratis-grpc/src/test/java/org/apache/ratis/grpc/MiniRaftClusterWithGRpc.java +++ b/ratis-grpc/src/test/java/org/apache/ratis/grpc/MiniRaftClusterWithGRpc.java @@ -64,17 +64,6 @@ public class MiniRaftClusterWithGRpc extends MiniRaftCluster.RpcBase { } @Override - protected void startServer(RaftServer server, boolean startService) { - final String id = server.getId().toString(); - if (startService) { - server.start(); - BlockRequestHandlingInjection.getInstance().unblockReplier(id); - } else { - BlockRequestHandlingInjection.getInstance().blockReplier(id); - } - } - - @Override protected void blockQueueAndSetDelay(String leaderId, int delayMs) throws InterruptedException { RaftTestUtil.blockQueueAndSetDelay(getServers(), sendServerRequestInjection, http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java b/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java index dcdcee4..cc8c8fb 100644 --- a/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java +++ b/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java @@ -17,13 +17,17 @@ */ package org.apache.ratis.grpc; +import org.apache.ratis.BaseTest; import org.apache.ratis.RaftTestUtil; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.protocol.RaftPeerId; import org.apache.ratis.server.impl.RaftServerTestUtil; import org.junit.Test; -public class TestRaftServerWithGrpc { +import java.io.IOException; +import java.nio.channels.OverlappingFileLockException; + +public class TestRaftServerWithGrpc extends BaseTest { @Test public void testServerRestartOnException() throws Exception { @@ -40,16 +44,15 @@ public class TestRaftServerWithGrpc { new RaftProperties(), null); // Close the server rpc for leader so that new raft server can be bound to it. cluster.getLeader().getServerRpc().close(); - try { - // Create a raft server proxy with server rpc bound to same address as - // the leader. This step would fail as the raft storage has been locked by - // the raft server proxy created earlier. Raft server proxy should close - // the rpc server on failure. - RaftServerTestUtil - .getRaftServerProxy(leaderId, cluster.getLeader().getStateMachine(), cluster.getGroup(), - properties, null); - } catch (Exception e) { - } + + // Create a raft server proxy with server rpc bound to same address as + // the leader. This step would fail as the raft storage has been locked by + // the raft server proxy created earlier. Raft server proxy should close + // the rpc server on failure. + testFailureCase("start a new server with the same address", + () -> RaftServerTestUtil.getRaftServerProxy(leaderId, cluster.getLeader().getStateMachine(), + cluster.getGroup(), properties, null), + IOException.class, OverlappingFileLockException.class); // Try to start a raft server rpc at the leader address. cluster.getServer(leaderId).getFactory().newRaftServerRpc(cluster.getServer(leaderId)); } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java index a6adf8c..3e9acfa 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java @@ -28,11 +28,7 @@ import org.apache.ratis.server.storage.RaftLogIOException; import org.apache.ratis.shaded.com.google.protobuf.ByteString; import org.apache.ratis.shaded.proto.RaftProtos.*; import org.apache.ratis.statemachine.SnapshotInfo; -import org.apache.ratis.util.Daemon; -import org.apache.ratis.util.IOUtils; -import org.apache.ratis.util.Preconditions; -import org.apache.ratis.util.ProtoUtils; -import org.apache.ratis.util.Timestamp; +import org.apache.ratis.util.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -60,6 +56,7 @@ public class LogAppender extends Daemon { private final boolean batchSending; private final LogEntryBuffer buffer; private final int snapshotChunkMaxSize; + protected final long halfMinTimeoutMs; private volatile boolean sending = true; @@ -73,6 +70,7 @@ public class LogAppender extends Daemon { this.maxBufferSize = RaftServerConfigKeys.Log.Appender.bufferCapacity(properties).getSizeInt(); this.batchSending = RaftServerConfigKeys.Log.Appender.batchEnabled(properties); this.snapshotChunkMaxSize = RaftServerConfigKeys.Log.Appender.snapshotChunkSizeMax(properties).getSizeInt(); + this.halfMinTimeoutMs = server.getMinTimeoutMs() / 2; this.buffer = new LogEntryBuffer(); } @@ -157,8 +155,8 @@ public class LogAppender extends Daemon { // if previous is null, nextIndex must be equal to the log start // index (otherwise we will install snapshot). Preconditions.assertTrue(follower.getNextIndex() == raftLog.getStartIndex(), - "follower's next index %s, local log start index %s", - follower.getNextIndex(), raftLog.getStartIndex()); + "%s: follower's next index %s, local log start index %s", + this, follower.getNextIndex(), raftLog.getStartIndex()); SnapshotInfo snapshot = server.getState().getLatestSnapshot(); previous = snapshot == null ? null : snapshot.getTermIndex(); } @@ -343,8 +341,7 @@ public class LogAppender extends Daemon { return builder.build(); } - private InstallSnapshotReplyProto installSnapshot(SnapshotInfo snapshot) - throws InterruptedException, InterruptedIOException { + private InstallSnapshotReplyProto installSnapshot(SnapshotInfo snapshot) throws InterruptedIOException { String requestId = UUID.randomUUID().toString(); InstallSnapshotReplyProto reply = null; try { @@ -415,8 +412,7 @@ public class LogAppender extends Daemon { } if (isAppenderRunning() && !shouldAppendEntries( follower.getNextIndex() + buffer.getPendingEntryNum())) { - final long waitTime = getHeartbeatRemainingTime( - follower.getLastRpcTime()); + final long waitTime = getHeartbeatRemainingTime(); if (waitTime > 0) { synchronized (this) { wait(waitTime); @@ -480,14 +476,14 @@ public class LogAppender extends Daemon { } private boolean shouldHeartbeat() { - return getHeartbeatRemainingTime(follower.getLastRpcTime()) <= 0; + return getHeartbeatRemainingTime() <= 0; } /** * @return the time in milliseconds that the leader should send a heartbeat. */ - protected long getHeartbeatRemainingTime(Timestamp lastTime) { - return server.getMinTimeoutMs() / 2 - lastTime.elapsedTimeMs(); + protected long getHeartbeatRemainingTime() { + return halfMinTimeoutMs - follower.getLastRpcTime().elapsedTimeMs(); } protected void checkResponseTerm(long responseTerm) { http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java index 28ce59f..6d1eb96 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java @@ -73,6 +73,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou private final LifeCycle lifeCycle; private final ServerState state; private final RaftGroupId groupId; + private final Supplier<RaftPeer> peerSupplier = JavaUtils.memoize(() -> new RaftPeer(getId(), getServerRpc().getInetSocketAddress())); private volatile Role role; /** used when the peer is follower, to monitor election timeout */ @@ -100,8 +101,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou "max timeout: %s, min timeout: %s", maxTimeoutMs, minTimeoutMs); this.proxy = proxy; - final RaftPeer peer = new RaftPeer(id, proxy.getServerRpc().getInetSocketAddress()); - this.state = new ServerState(peer, group, properties, this, proxy.getStateMachine()); + this.state = new ServerState(id, group, properties, this, proxy.getStateMachine()); this.retryCache = initRetryCache(properties); this.jmxAdapter = new RaftServerJmxAdapter(); @@ -120,6 +120,9 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou return getProxy().getFactory().newLogAppender(this, state, f); } + RaftPeer getPeer() { + return peerSupplier.get(); + } int getMinTimeoutMs() { return minTimeoutMs; @@ -346,7 +349,7 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou Collection<CommitInfoProto> getCommitInfos() { final List<CommitInfoProto> infos = new ArrayList<>(); // add the commit info of this server - infos.add(state.updateCommitInfo(commitInfoCache)); + infos.add(commitInfoCache.update(getPeer(), state.getLog().getLastCommittedIndex())); // add the commit infos of other servers if (isLeader()) { http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java index d7b7ad0..cfb369b 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerProxy.java @@ -62,12 +62,14 @@ public class RaftServerProxy implements RaftServer { this.serverRpc = factory.newRaftServerRpc(this); this.id = id != null? id: RaftPeerId.valueOf(getIdStringFrom(serverRpc)); + try { this.impl = CompletableFuture.completedFuture(initImpl(group)); } catch (IOException ioe) { try { serverRpc.close(); } catch (IOException closeIoe) { + LOG.warn(this.id + ": Failed to close server rpc.", closeIoe); ioe.addSuppressed(closeIoe); } finally { throw ioe; @@ -80,7 +82,12 @@ public class RaftServerProxy implements RaftServer { } private static String getIdStringFrom(RaftServerRpc rpc) { - final InetSocketAddress address = rpc.getInetSocketAddress(); + InetSocketAddress address = null; + try { + address = rpc.getInetSocketAddress(); + } catch(Exception e) { + LOG.warn("Failed to get InetSocketAddress from " + rpc.getRpcType() + " rpc server", e); + } return address != null? address.getHostName() + "_" + address.getPort() : rpc.getRpcType() + "-" + UUID.randomUUID(); } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java index 5d91120..8c7b269 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ServerState.java @@ -44,7 +44,6 @@ import static org.apache.ratis.shaded.proto.RaftProtos.LogEntryProto.LogEntryBod */ public class ServerState implements Closeable { private final RaftPeerId selfId; - private final RaftPeer peer; private final RaftServerImpl server; /** Raft log */ private final RaftLog log; @@ -78,11 +77,10 @@ public class ServerState implements Closeable { */ private TermIndex latestInstalledSnapshot; - ServerState(RaftPeer peer, RaftGroup group, RaftProperties prop, + ServerState(RaftPeerId id, RaftGroup group, RaftProperties prop, RaftServerImpl server, StateMachine stateMachine) throws IOException { - this.selfId = peer.getId(); - this.peer = peer; + this.selfId = id; this.server = server; RaftConfiguration initialConf = RaftConfiguration.newBuilder() .setConf(group.getPeers()).build(); @@ -91,14 +89,14 @@ public class ServerState implements Closeable { final File dir = RaftServerConfigKeys.storageDir(prop); storage = new RaftStorage(new File(dir, group.getGroupId().toString()), RaftServerConstants.StartupOption.REGULAR); - snapshotManager = new SnapshotManager(storage, peer.getId()); + snapshotManager = new SnapshotManager(storage, id); long lastApplied = initStatemachine(stateMachine, prop); leaderId = null; // we cannot apply log entries to the state machine in this step, since we // do not know whether the local log entries have been committed. - log = initLog(peer.getId(), prop, lastApplied, entry -> { + log = initLog(id, prop, lastApplied, entry -> { if (entry.getLogEntryBodyCase() == CONFIGURATIONENTRY) { configurationManager.addConfiguration(entry.getIndex(), ServerProtoUtils.toRaftConfiguration(entry.getIndex(), @@ -165,10 +163,6 @@ public class ServerState implements Closeable { return this.selfId; } - CommitInfoProto updateCommitInfo(CommitInfoCache cache) { - return cache.update(peer, log.getLastCommittedIndex()); - } - public long getCurrentTerm() { return currentTerm; } @@ -253,8 +247,10 @@ public class ServerState implements Closeable { /** * Check if the candidate's term is acceptable */ - boolean recognizeCandidate(RaftPeerId candidateId, - long candidateTerm) { + boolean recognizeCandidate(RaftPeerId candidateId, long candidateTerm) { + if (!getRaftConf().containsInConf(candidateId)) { + return false; + } if (candidateTerm > currentTerm) { return true; } else if (candidateTerm == currentTerm) { http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java b/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java index ded7d52..24d78b7 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/storage/RaftStorageDirectory.java @@ -320,7 +320,7 @@ public class RaftStorageDirectory { LOG.error("It appears that another process " + "has already locked the storage directory: " + root, oe); file.close(); - return null; + throw new IOException("Failed to lock storage " + this.root + ". The directory is already locked", oe); } catch(IOException e) { LOG.error("Failed to acquire lock on " + lockF + ". If this storage directory is mounted via NFS, " http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java b/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java index 74238e8..5c5749f 100644 --- a/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java +++ b/ratis-server/src/test/java/org/apache/ratis/MiniRaftCluster.java @@ -274,14 +274,18 @@ public abstract class MiniRaftCluster { return addNewPeers(generateIds(number, servers.size()), startNewPeer); } - public PeerChanges addNewPeers(String[] ids, - boolean startNewPeer) throws IOException { + public PeerChanges addNewPeers(String[] ids, boolean startNewPeer) { LOG.info("Add new peers {}", Arrays.asList(ids)); // create and add new RaftServers final Collection<RaftServerProxy> newServers = putNewServers( CollectionUtils.as(Arrays.asList(ids), RaftPeerId::valueOf), true); - newServers.forEach(s -> startServer(s, startNewPeer)); + + newServers.forEach(s -> startServer(s, true)); + if (!startNewPeer) { + // start and then close, in order to bind the port + newServers.forEach(p -> p.close()); + } final Collection<RaftPeer> newPeers = toRaftPeers(newServers); final RaftPeer[] np = newPeers.toArray(new RaftPeer[newPeers.size()]); http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/451e8551/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java index 9b142fd..1b0ef0c 100644 --- a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java +++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java @@ -272,7 +272,7 @@ public abstract class RaftReconfigurationBaseTest extends BaseTest { // start the two new peers LOG.info("Start new peers"); for (RaftPeer np : c1.newPeers) { - cluster.startServer(np.getId()); + cluster.restartServer(np.getId(), false); } Assert.assertTrue(client.setConfiguration(c1.allPeersInNewConf).isSuccess()); client.close(); @@ -380,7 +380,7 @@ public abstract class RaftReconfigurationBaseTest extends BaseTest { final String oldLeaderId = RaftTestUtil.waitAndKillLeader(cluster, true); LOG.info("start the two new peers: {}", Arrays.asList(c1.newPeers)); for (RaftPeer np : c1.newPeers) { - cluster.startServer(np.getId()); + cluster.restartServer(np.getId(), false); } Thread.sleep(3000); @@ -585,15 +585,15 @@ public abstract class RaftReconfigurationBaseTest extends BaseTest { public void testLeaderNotReadyException() throws Exception { LOG.info("Start testLeaderNotReadyException"); final MiniRaftCluster cluster = getCluster(1).initServers(); - final RaftPeerId leaderId = cluster.getPeers().iterator().next().getId(); try { // delay 1s for each logSync call - cluster.getPeers().forEach( + cluster.getServers().forEach( peer -> leaderPlaceHolderDelay.setDelayMs(peer.getId().toString(), 2000)); cluster.start(); AtomicBoolean caughtNotReady = new AtomicBoolean(false); AtomicBoolean success = new AtomicBoolean(false); + final RaftPeerId leaderId = cluster.getPeers().iterator().next().getId(); new Thread(() -> { final RaftClient client = cluster.createClient(leaderId); final RaftClientRpc sender = client.getClientRpc();
