Repository: incubator-ratis Updated Branches: refs/heads/master 7b3a9a6f5 -> 9f93037a9
RATIS-196. RaftServerProtocolService may continue to reply after it is closed. Contributed by Mukul Kumar Singh. Project: http://git-wip-us.apache.org/repos/asf/incubator-ratis/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ratis/commit/9f93037a Tree: http://git-wip-us.apache.org/repos/asf/incubator-ratis/tree/9f93037a Diff: http://git-wip-us.apache.org/repos/asf/incubator-ratis/diff/9f93037a Branch: refs/heads/master Commit: 9f93037a965970d0de1db7920ac498901886c45e Parents: 7b3a9a6 Author: Mukul Kumar Singh <[email protected]> Authored: Mon Jan 22 10:48:52 2018 +0530 Committer: Mukul Kumar Singh <[email protected]> Committed: Mon Jan 22 10:48:52 2018 +0530 ---------------------------------------------------------------------- .../ratis/grpc/server/RaftServerProtocolService.java | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/9f93037a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolService.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolService.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolService.java index 1e499ae..50e2343 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolService.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolService.java @@ -28,6 +28,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -66,6 +67,7 @@ public class RaftServerProtocolService extends RaftServerProtocolServiceImplBase return new StreamObserver<AppendEntriesRequestProto>() { private final AtomicReference<CompletableFuture<Void>> previousOnNext = new AtomicReference<>(CompletableFuture.completedFuture(null)); + private final AtomicBoolean isClosed = new AtomicBoolean(false); @Override public void onNext(AppendEntriesRequestProto request) { @@ -74,7 +76,9 @@ public class RaftServerProtocolService extends RaftServerProtocolServiceImplBase try { server.appendEntriesAsync(request).thenCombine(previous, (reply, v) -> { - responseObserver.onNext(reply); + if (!isClosed.get()) { + responseObserver.onNext(reply); + } current.complete(null); return null; }); @@ -96,8 +100,10 @@ public class RaftServerProtocolService extends RaftServerProtocolServiceImplBase @Override public void onCompleted() { - LOG.info("{}: appendEntries completed", getId()); - responseObserver.onCompleted(); + if (isClosed.compareAndSet(false, true)) { + LOG.info("{}: appendEntries completed", getId()); + responseObserver.onCompleted(); + } } }; }
