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 2021/12/09 07:47:15 UTC
[ratis] branch master updated: RATIS-1457. DataStreamManagement startTransaction error handler (#552)
This is an automated email from the ASF dual-hosted git repository.
szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ratis.git
The following commit(s) were added to refs/heads/master by this push:
new 2faf872 RATIS-1457. DataStreamManagement startTransaction error handler (#552)
2faf872 is described below
commit 2faf8721265ae2f9c3551c0d487ae50619e19ee0
Author: hao guo <gu...@360.cn>
AuthorDate: Thu Dec 9 15:47:06 2021 +0800
RATIS-1457. DataStreamManagement startTransaction error handler (#552)
---
.../apache/ratis/netty/server/DataStreamManagement.java | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java
index 98e9c9e..81621c3 100644
--- a/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java
+++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java
@@ -323,16 +323,20 @@ public class DataStreamManagement {
ctx.writeAndFlush(builder.build());
}
- private CompletableFuture<Void> startTransaction(StreamInfo info, DataStreamRequestByteBuf request, long bytesWritten,
- ChannelHandlerContext ctx) {
+ private CompletableFuture<RaftClientReply> startTransaction(StreamInfo info, DataStreamRequestByteBuf request,
+ long bytesWritten, ChannelHandlerContext ctx) {
try {
AsyncRpcApi asyncRpcApi = (AsyncRpcApi) (server.getDivision(info.getRequest()
.getRaftGroupId())
.getRaftClient()
.async());
- return asyncRpcApi.sendForward(info.request).thenAcceptAsync(
- reply -> ctx.writeAndFlush(newDataStreamReplyByteBuffer(request, reply, bytesWritten, info.getCommitInfos())),
- requestExecutor);
+ return asyncRpcApi.sendForward(info.request).whenCompleteAsync((reply, e) -> {
+ if (e != null) {
+ replyDataStreamException(server, e, info.getRequest(), request, ctx);
+ } else {
+ ctx.writeAndFlush(newDataStreamReplyByteBuffer(request, reply, bytesWritten, info.getCommitInfos()));
+ }
+ }, requestExecutor);
} catch (IOException e) {
throw new CompletionException(e);
}