You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by sz...@apache.org on 2018/04/30 17:51:04 UTC
incubator-ratis git commit: RATIS-231. Grpc server should reconnect
if there is an exception.
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 <sz...@HW14119.local>
Authored: Mon Apr 30 10:50:03 2018 -0700
Committer: Tsz Sze <sz...@HW14119.local>
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();