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 fd2dde635 RATIS-2564. Add an executor and terminalReply to stream read
(#1486)
fd2dde635 is described below
commit fd2dde635c02ffc697eb7e782a2cbd4d10234778
Author: Peter Lee <[email protected]>
AuthorDate: Fri Jun 19 02:09:10 2026 +0800
RATIS-2564. Add an executor and terminalReply to stream read (#1486)
---
.../apache/ratis/client/impl/ClientProtoUtils.java | 11 ++--
.../ratis/client/impl/DataStreamClientImpl.java | 23 ++++----
.../datastream/impl/DataStreamReplyByteBuf.java | 24 +++++++-
.../apache/ratis/protocol/ClientInvocationId.java | 4 ++
.../java/org/apache/ratis/protocol/Message.java | 5 ++
.../java/org/apache/ratis/util/NettyUtils.java | 7 +++
.../apache/ratis/netty/NettyDataStreamUtils.java | 30 +++++-----
.../ratis/netty/client/NettyClientReplies.java | 14 ++---
.../ratis/netty/client/NettyClientStreamRpc.java | 16 +++--
.../ratis/netty/server/DataStreamManagement.java | 2 +-
.../ratis/netty/server/NettyServerStreamRpc.java | 22 +++++--
.../ratis/netty/server/ReadStreamManagement.java | 68 ++++++++++++++++++----
.../netty/server/TestDataStreamManagement.java | 4 +-
13 files changed, 170 insertions(+), 60 deletions(-)
diff --git
a/ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java
b/ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java
index d2146a521..1126453c2 100644
---
a/ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java
+++
b/ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java
@@ -18,6 +18,7 @@
package org.apache.ratis.client.impl;
import org.apache.ratis.datastream.impl.DataStreamReplyByteBuffer;
+import org.apache.ratis.datastream.impl.DataStreamReplyByteBuf;
import org.apache.ratis.proto.RaftProtos.AlreadyClosedExceptionProto;
import org.apache.ratis.proto.RaftProtos.ClientMessageEntryProto;
import org.apache.ratis.proto.RaftProtos.GroupAddRequestProto;
@@ -379,11 +380,13 @@ public interface ClientProtoUtils {
}
static RaftClientReply getRaftClientReply(DataStreamReply reply) {
- if (!(reply instanceof DataStreamReplyByteBuffer)) {
- throw new IllegalStateException("Unexpected " + reply.getClass() + ":
reply is " + reply);
- }
try {
- return toRaftClientReply(((DataStreamReplyByteBuffer) reply).slice());
+ if (reply instanceof DataStreamReplyByteBuffer) {
+ return toRaftClientReply(((DataStreamReplyByteBuffer) reply).slice());
+ } else if (reply instanceof DataStreamReplyByteBuf) {
+ return toRaftClientReply(((DataStreamReplyByteBuf)
reply).slice().nioBuffer());
+ }
+ throw new IllegalStateException("Unexpected " + reply.getClass() + ":
reply is " + reply);
} catch (InvalidProtocolBufferException e) {
throw new IllegalStateException("Failed to getRaftClientReply from " +
reply, e);
}
diff --git
a/ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
b/ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
index 313131cbd..3c055009f 100644
---
a/ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
+++
b/ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
@@ -34,16 +34,16 @@ import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.protocol.ClientInvocationId;
import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.protocol.DataStreamRequestHeader;
+import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeer;
+import org.apache.ratis.protocol.RoutingTable;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.rpc.CallId;
import org.apache.ratis.thirdparty.io.netty.buffer.ByteBuf;
import org.apache.ratis.util.IOUtils;
-import org.apache.ratis.protocol.*;
-import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
import org.apache.ratis.util.JavaUtils;
import org.apache.ratis.util.MemoizedSupplier;
import org.apache.ratis.util.Preconditions;
@@ -57,7 +57,6 @@ import java.nio.channels.WritableByteChannel;
import java.util.Arrays;
import java.util.Collections;
import java.util.Objects;
-import java.util.Optional;
import java.util.concurrent.CompletableFuture;
/**
@@ -260,18 +259,20 @@ public class DataStreamClientImpl implements
DataStreamClient {
() -> "Primary peer mismatched: the routing table has " +
routingTable.getPrimary()
+ " but the client has " + dataStreamServer.getId());
}
- final Message message =
-
Optional.ofNullable(headerMessage).map(ByteString::copyFrom).map(Message::valueOf).orElse(null);
- RaftClientRequest request = RaftClientRequest.newBuilder()
+ return new DataStreamOutputImpl(newBuilder(headerMessage)
+ .setType(RaftClientRequest.dataStreamRequestType())
+ .setRoutingTable(routingTable)
+ .build());
+ }
+
+ private RaftClientRequest.Builder newBuilder(ByteBuffer headerMessage) {
+ final Message message = headerMessage == null ? null :
Message.valueOf(headerMessage);
+ return RaftClientRequest.newBuilder()
.setClientId(clientId)
.setServerId(dataStreamServer.getId())
.setGroupId(groupId)
.setCallId(CallId.getAndIncrement())
- .setMessage(message)
- .setType(RaftClientRequest.dataStreamRequestType())
- .setRoutingTable(routingTable)
- .build();
- return new DataStreamOutputImpl(request);
+ .setMessage(message);
}
@Override
diff --git
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuf.java
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuf.java
index bef58563f..2477b089b 100644
---
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuf.java
+++
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuf.java
@@ -22,6 +22,7 @@ import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.protocol.DataStreamReplyHeader;
import org.apache.ratis.thirdparty.io.netty.buffer.ByteBuf;
+import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Collections;
@@ -30,7 +31,7 @@ import java.util.Collections;
* <p>
* This class is immutable.
*/
-public final class DataStreamReplyByteBuf extends DataStreamPacketByteBuf
implements DataStreamReply {
+public final class DataStreamReplyByteBuf extends DataStreamPacketByteBuf
implements DataStreamReply, AutoCloseable {
public static final class Builder extends DataStreamReplyBuilder<Builder> {
private ByteBuf buf;
@@ -90,10 +91,31 @@ public final class DataStreamReplyByteBuf extends
DataStreamPacketByteBuf implem
return Collections.unmodifiableCollection(commitInfos);
}
+ @Override
+ public void close() {
+ release();
+ }
+
@Override
public String toString() {
return super.toString()
+ "," + (success ? "SUCCESS" : "FAILED")
+ ",bytesWritten=" + bytesWritten;
}
+
+ public DataStreamReplyByteBuffer copy() {
+ return DataStreamReplyByteBuffer.newBuilder()
+ .setDataStreamPacket(this)
+ .setBuffer(copy(slice()))
+ .setSuccess(isSuccess())
+ .setBytesWritten(getBytesWritten())
+ .setCommitInfos(getCommitInfos())
+ .build();
+ }
+
+ static ByteBuffer copy(ByteBuf buf) {
+ final byte[] bytes = new byte[buf.readableBytes()];
+ buf.readBytes(bytes);
+ return ByteBuffer.wrap(bytes);
+ }
}
diff --git
a/ratis-common/src/main/java/org/apache/ratis/protocol/ClientInvocationId.java
b/ratis-common/src/main/java/org/apache/ratis/protocol/ClientInvocationId.java
index 24e491e6e..0d05b4c92 100644
---
a/ratis-common/src/main/java/org/apache/ratis/protocol/ClientInvocationId.java
+++
b/ratis-common/src/main/java/org/apache/ratis/protocol/ClientInvocationId.java
@@ -37,6 +37,10 @@ public final class ClientInvocationId {
return valueOf(message.getClientId(), message.getCallId());
}
+ public static ClientInvocationId valueOf(DataStreamPacket packet) {
+ return valueOf(packet.getClientId(), packet.getStreamId());
+ }
+
public static ClientInvocationId valueOf(StateMachineLogEntryProto proto) {
return valueOf(ClientId.valueOf(proto.getClientId()), proto.getCallId());
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/protocol/Message.java
b/ratis-common/src/main/java/org/apache/ratis/protocol/Message.java
index 55fcd064d..b773826fb 100644
--- a/ratis-common/src/main/java/org/apache/ratis/protocol/Message.java
+++ b/ratis-common/src/main/java/org/apache/ratis/protocol/Message.java
@@ -23,6 +23,7 @@ import
org.apache.ratis.thirdparty.com.google.protobuf.TextFormat;
import org.apache.ratis.util.MemoizedSupplier;
import org.apache.ratis.util.StringUtils;
+import java.nio.ByteBuffer;
import java.util.Optional;
import java.util.function.Supplier;
@@ -55,6 +56,10 @@ public interface Message {
return valueOf(bytes, () -> "Message:" +
StringUtils.bytes2ShortString(bytes));
}
+ static Message valueOf(ByteBuffer bytes) {
+ return valueOf(ByteString.copyFrom(bytes));
+ }
+
static Message valueOf(String string) {
return valueOf(ByteString.copyFromUtf8(string), () -> "Message:" + string);
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/NettyUtils.java
b/ratis-common/src/main/java/org/apache/ratis/util/NettyUtils.java
index 163c3d869..819c93db9 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/NettyUtils.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/NettyUtils.java
@@ -37,6 +37,7 @@ import
org.apache.ratis.thirdparty.io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslContext;
import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslContextBuilder;
import org.apache.ratis.thirdparty.io.netty.util.concurrent.Future;
+import org.apache.ratis.thirdparty.io.netty.util.concurrent.ScheduledFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -230,4 +231,10 @@ public interface NettyUtils {
LOG.warn("closeChannel {} is not yet completed in {}", name,
CLOSE_TIMEOUT);
}
}
+
+ static void cancel(ScheduledFuture<?> future) {
+ if (future != null) {
+ future.cancel(true);
+ }
+ }
}
diff --git
a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java
b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java
index 583d6e3e9..2c5015b12 100644
--- a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java
+++ b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java
@@ -18,6 +18,7 @@
package org.apache.ratis.netty;
import org.apache.ratis.datastream.impl.DataStreamReplyByteBuffer;
+import org.apache.ratis.datastream.impl.DataStreamReplyByteBuf;
import org.apache.ratis.datastream.impl.DataStreamRequestByteBuffer;
import org.apache.ratis.datastream.impl.DataStreamRequestFilePositionCount;
import org.apache.ratis.io.FilePositionCount;
@@ -30,6 +31,7 @@ import
org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto;
import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.protocol.DataStreamPacketHeader;
import org.apache.ratis.protocol.DataStreamReplyHeader;
+import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.protocol.DataStreamRequest;
import org.apache.ratis.protocol.DataStreamRequestHeader;
import
org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
@@ -80,7 +82,7 @@ public interface NettyDataStreamUtils {
.asReadOnlyByteBuffer();
}
- static ByteBuffer
getDataStreamReplyHeaderProtoByteBuf(DataStreamReplyByteBuffer reply) {
+ static ByteBuffer getDataStreamReplyHeaderProtoByteBuf(DataStreamReply
reply) {
DataStreamPacketHeaderProto.Builder b = DataStreamPacketHeaderProto
.newBuilder()
.setClientId(reply.getClientId().toByteString())
@@ -149,14 +151,22 @@ public interface NettyDataStreamUtils {
out.accept(new DefaultFileRegion(f.getFile(), f.getPosition(),
f.getCount()));
}
- static void encodeDataStreamReplyByteBuffer(DataStreamReplyByteBuffer reply,
Consumer<ByteBuf> out,
+ static void encodeDataStreamReply(DataStreamReply reply, ByteBuf body,
Consumer<Object> out,
ByteBufAllocator allocator) {
- ByteBuffer headerBuf = getDataStreamReplyHeaderProtoByteBuf(reply);
+ final ByteBuffer headerBuf = getDataStreamReplyHeaderProtoByteBuf(reply);
final ByteBuf headerLenBuf =
allocator.ioBuffer(DataStreamPacketHeader.getSizeOfHeaderLen());
headerLenBuf.writeInt(headerBuf.remaining());
out.accept(headerLenBuf);
out.accept(Unpooled.wrappedBuffer(headerBuf));
- out.accept(Unpooled.wrappedBuffer(reply.slice()));
+ encodeByteBuf(body, out);
+ }
+
+ static void encodeDataStreamReply(DataStreamReplyByteBuffer reply,
Consumer<Object> out, ByteBufAllocator allocator) {
+ encodeDataStreamReply(reply, Unpooled.wrappedBuffer(reply.slice()), out,
allocator);
+ }
+
+ static void encodeDataStreamReply(DataStreamReplyByteBuf reply,
Consumer<Object> out, ByteBufAllocator allocator) {
+ encodeDataStreamReply(reply, reply.slice(), out, allocator);
}
static DataStreamRequestByteBuf decodeDataStreamRequestByteBuf(ByteBuf buf) {
@@ -208,18 +218,12 @@ public interface NettyDataStreamUtils {
}
}
- static ByteBuffer copy(ByteBuf buf) {
- final byte[] bytes = new byte[buf.readableBytes()];
- buf.readBytes(bytes);
- return ByteBuffer.wrap(bytes);
- }
-
- static DataStreamReplyByteBuffer decodeDataStreamReplyByteBuffer(ByteBuf
buf) {
+ static DataStreamReplyByteBuf decodeDataStreamReplyByteBuf(ByteBuf buf) {
return Optional.ofNullable(decodeDataStreamReplyHeader(buf))
.map(header -> checkHeader(header, buf))
- .map(header -> DataStreamReplyByteBuffer.newBuilder()
+ .map(header -> DataStreamReplyByteBuf.newBuilder()
.setDataStreamReplyHeader(header)
- .setBuffer(decodeData(buf, header, NettyDataStreamUtils::copy))
+ .setBuf(decodeData(buf, header, ByteBuf::retainedSlice))
.build())
.orElse(null);
}
diff --git
a/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientReplies.java
b/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientReplies.java
index 695177262..121ef3c0f 100644
---
a/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientReplies.java
+++
b/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientReplies.java
@@ -18,12 +18,14 @@
package org.apache.ratis.netty.client;
+import org.apache.ratis.datastream.impl.DataStreamReplyByteBuf;
import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import org.apache.ratis.protocol.ClientInvocationId;
import org.apache.ratis.protocol.DataStreamPacket;
import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.thirdparty.io.netty.util.concurrent.ScheduledFuture;
import org.apache.ratis.util.MemoizedSupplier;
+import org.apache.ratis.util.NettyUtils;
import org.apache.ratis.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,7 +65,7 @@ public class NettyClientReplies {
return map.computeIfAbsent(requestEntry, r -> new ReplyEntry(isClose,
f));
}
- void receiveReply(DataStreamReply reply) {
+ void receiveReply(DataStreamReplyByteBuf reply) {
final RequestEntry requestEntry = new RequestEntry(reply);
final ReplyEntry replyEntry = map.remove(requestEntry);
LOG.debug("remove: {}; replyEntry: {}; reply: {}", requestEntry,
replyEntry, reply);
@@ -162,21 +164,15 @@ public class NettyClientReplies {
}
synchronized void complete(DataStreamReply reply) {
- cancel(timeoutFuture);
+ NettyUtils.cancel(timeoutFuture);
replyFuture.complete(reply);
}
synchronized void completeExceptionally(Throwable t) {
- cancel(timeoutFuture);
+ NettyUtils.cancel(timeoutFuture);
replyFuture.completeExceptionally(t);
}
- static void cancel(ScheduledFuture<?> future) {
- if (future != null) {
- future.cancel(true);
- }
- }
-
synchronized void scheduleTimeout(Supplier<ScheduledFuture<?>>
scheduleMethod) {
if (!replyFuture.isDone()) {
timeoutFuture = scheduleMethod.get();
diff --git
a/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java
b/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java
index 5b673d518..ccb0b15da 100644
---
a/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java
+++
b/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java
@@ -23,6 +23,7 @@ import org.apache.ratis.client.RaftClientConfigKeys;
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.datastream.impl.DataStreamRequestByteBuf;
import org.apache.ratis.datastream.impl.DataStreamRequestByteBuffer;
+import org.apache.ratis.datastream.impl.DataStreamReplyByteBuf;
import org.apache.ratis.datastream.impl.DataStreamRequestFilePositionCount;
import org.apache.ratis.io.StandardWriteOption;
import org.apache.ratis.io.WriteOption;
@@ -353,11 +354,16 @@ public class NettyClientStreamRpc implements
DataStreamClientRpc {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
- if (!(msg instanceof DataStreamReply)) {
+ if (!(msg instanceof DataStreamReplyByteBuf)) {
LOG.error("{}: unexpected message {}", name, msg.getClass());
return;
}
- final DataStreamReply reply = (DataStreamReply) msg;
+ try (DataStreamReplyByteBuf reply = (DataStreamReplyByteBuf) msg) {
+ process(reply);
+ }
+ }
+
+ private void process(DataStreamReplyByteBuf reply) {
LOG.debug("{}: read {}", name, reply);
final ClientInvocationId clientInvocationId =
ClientInvocationId.valueOf(
reply.getClientId(), reply.getStreamId());
@@ -370,7 +376,7 @@ public class NettyClientStreamRpc implements
DataStreamClientRpc {
try {
replyMap.receiveReply(reply);
} catch (Throwable cause) {
- LOG.warn("{} : channelRead error:", name, cause);
+ LOG.warn("{} : channelRead error for {}", name, reply, cause);
replyMap.completeExceptionally(cause);
}
}
@@ -456,7 +462,7 @@ public class NettyClientStreamRpc implements
DataStreamClientRpc {
@Override
protected void decode(ChannelHandlerContext context, ByteBuf buf,
List<Object> out) {
-
Optional.ofNullable(NettyDataStreamUtils.decodeDataStreamReplyByteBuffer(buf)).ifPresent(out::add);
+
Optional.ofNullable(NettyDataStreamUtils.decodeDataStreamReplyByteBuf(buf)).ifPresent(out::add);
}
};
}
@@ -464,7 +470,7 @@ public class NettyClientStreamRpc implements
DataStreamClientRpc {
@Override
public CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest
request) {
final CompletableFuture<DataStreamReply> f = new CompletableFuture<>();
- ClientInvocationId clientInvocationId =
ClientInvocationId.valueOf(request.getClientId(), request.getStreamId());
+ final ClientInvocationId clientInvocationId =
ClientInvocationId.valueOf(request);
final boolean isClose =
request.getWriteOptionList().contains(StandardWriteOption.CLOSE);
final NettyClientReplies.ReplyMap replyMap =
replies.getOrCreateReplyMap(clientInvocationId);
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 4e337e06b..005e0b8e6 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
@@ -424,7 +424,7 @@ public class DataStreamManagement {
void read(DataStreamRequestByteBuf request, ChannelHandlerContext ctx,
CheckedBiFunction<RaftClientRequest, Set<RaftPeer>,
Set<DataStreamOutputImpl>, IOException> getStreams) {
LOG.debug("{}: read {}", this, request);
- final ClientInvocationId key =
ClientInvocationId.valueOf(request.getClientId(), request.getStreamId());
+ final ClientInvocationId key = ClientInvocationId.valueOf(request);
final ChannelId channelId = ctx.channel().id();
try {
readImpl(request, ctx, getStreams, key, channelId);
diff --git
a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyServerStreamRpc.java
b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyServerStreamRpc.java
index 643ed15a0..d7cdeb056 100644
---
a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyServerStreamRpc.java
+++
b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyServerStreamRpc.java
@@ -23,6 +23,7 @@ import org.apache.ratis.client.DataStreamOutputRpc;
import org.apache.ratis.client.impl.DataStreamClientImpl.DataStreamOutputImpl;
import org.apache.ratis.conf.Parameters;
import org.apache.ratis.conf.RaftProperties;
+import org.apache.ratis.datastream.impl.DataStreamReplyByteBuf;
import org.apache.ratis.datastream.impl.DataStreamReplyByteBuffer;
import org.apache.ratis.datastream.impl.DataStreamRequestByteBuf;
import org.apache.ratis.netty.NettyConfigKeys;
@@ -31,6 +32,7 @@ import org.apache.ratis.util.NettyUtils;
import org.apache.ratis.netty.metrics.NettyServerStreamRpcMetrics;
import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.protocol.DataStreamPacket;
+import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftPeer;
import org.apache.ratis.security.TlsConf;
@@ -286,13 +288,19 @@ public class NettyServerStreamRpc implements
DataStreamServerRpc {
};
}
- static final MessageToMessageEncoder<DataStreamReplyByteBuffer> ENCODER =
new Encoder();
+ static final MessageToMessageEncoder<DataStreamReply> ENCODER = new
Encoder();
@ChannelHandler.Sharable
- static class Encoder extends
MessageToMessageEncoder<DataStreamReplyByteBuffer> {
+ static class Encoder extends MessageToMessageEncoder<DataStreamReply> {
@Override
- protected void encode(ChannelHandlerContext context,
DataStreamReplyByteBuffer reply, List<Object> out) {
- NettyDataStreamUtils.encodeDataStreamReplyByteBuffer(reply, out::add,
context.alloc());
+ protected void encode(ChannelHandlerContext context, DataStreamReply
reply, List<Object> out) {
+ if (reply instanceof DataStreamReplyByteBuffer) {
+ NettyDataStreamUtils.encodeDataStreamReply((DataStreamReplyByteBuffer)
reply, out::add, context.alloc());
+ } else if (reply instanceof DataStreamReplyByteBuf) {
+ NettyDataStreamUtils.encodeDataStreamReply((DataStreamReplyByteBuf)
reply, out::add, context.alloc());
+ } else {
+ throw new IllegalArgumentException("Unexpected DataStreamReply class "
+ reply.getClass());
+ }
}
}
@@ -321,6 +329,12 @@ public class NettyServerStreamRpc implements
DataStreamServerRpc {
LOG.error(this + ": Failed to shutdown request service.", e);
}
+ try {
+ reads.shutdown();
+ } catch (Exception e) {
+ LOG.error(this + ": Failed to shutdown read service.", e);
+ }
+
try {
channelFuture.channel().close().sync();
bossGroup.shutdownGracefully(0, 100, TimeUnit.MILLISECONDS);
diff --git
a/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
b/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
index bcdced1ea..5336760a0 100644
---
a/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
+++
b/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
@@ -17,19 +17,24 @@
*/
package org.apache.ratis.netty.server;
+import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.datastream.impl.DataStreamReplyByteBuffer;
import org.apache.ratis.datastream.impl.DataStreamRequestByteBuf;
import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto;
import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto.TypeCase;
import org.apache.ratis.protocol.ClientId;
+import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.RaftServerConfigKeys;
import
org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
import org.apache.ratis.thirdparty.io.netty.channel.ChannelFuture;
import org.apache.ratis.thirdparty.io.netty.channel.ChannelHandlerContext;
+import org.apache.ratis.util.ConcurrentUtils;
import org.apache.ratis.util.JavaUtils;
+import org.apache.ratis.util.TimeDuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,8 +43,10 @@ import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import static
org.apache.ratis.client.impl.ClientProtoUtils.toRaftClientRequest;
+import static
org.apache.ratis.client.impl.ClientProtoUtils.toRaftClientReplyProto;
import static
org.apache.ratis.netty.server.DataStreamManagement.replyDataStreamException;
public class ReadStreamManagement {
@@ -50,12 +57,27 @@ public class ReadStreamManagement {
private final long streamId;
private final ChannelHandlerContext ctx;
private final CompletableFuture<Void> closed = new CompletableFuture<>();
+ private final DataStreamReplyByteBuffer terminalReply;
private long streamOffset;
- ReadStream(DataStreamRequestByteBuf request, ChannelHandlerContext ctx) {
+ ReadStream(RaftClientRequest request, long streamId, ChannelHandlerContext
ctx) {
this.clientId = request.getClientId();
- this.streamId = request.getStreamId();
+ this.streamId = streamId;
this.ctx = ctx;
+
+ final RaftClientReply reply = RaftClientReply.newBuilder()
+ .setRequest(request)
+ .setSuccess()
+ .build();
+ this.terminalReply = DataStreamReplyByteBuffer.newBuilder()
+ .setClientId(clientId)
+ .setType(Type.STREAM_HEADER)
+ .setStreamId(streamId)
+ .setStreamOffset(0)
+
.setBuffer(toRaftClientReplyProto(reply).toByteString().asReadOnlyByteBuffer())
+ .setSuccess(true)
+ .setBytesWritten(0)
+ .build();
}
@Override
@@ -64,8 +86,10 @@ public class ReadStreamManagement {
}
@Override
- public void close() {
- closed.complete(null);
+ public synchronized void close() throws IOException {
+ if (closed.complete(null)) {
+ writeAndFlush(terminalReply, 0);
+ }
}
@Override
@@ -76,19 +100,23 @@ public class ReadStreamManagement {
buffer = buffer.asReadOnlyBuffer();
final int length = buffer.remaining();
final DataStreamReplyByteBuffer reply = newReply(buffer);
+ writeAndFlush(reply, length);
+ streamOffset += length;
+ return length;
+ }
+
+ private synchronized void writeAndFlush(DataStreamReplyByteBuffer reply,
int length) throws IOException {
+ final long offset = reply.getStreamOffset();
final ChannelFuture future = ctx.writeAndFlush(reply);
try {
future.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- throw new InterruptedIOException(
- "Interrupted while writing " + length + " bytes at offset " +
streamOffset);
+ throw new InterruptedIOException("Interrupted while writing " + length
+ " bytes at offset " + offset);
}
if (!future.isSuccess()) {
- throw new IOException("Failed to write " + length + " bytes at offset
" + streamOffset, future.cause());
+ throw new IOException("Failed to write " + length + " bytes at offset
" + offset, future.cause());
}
- streamOffset += length;
- return length;
}
private synchronized DataStreamReplyByteBuffer newReply(ByteBuffer buffer)
{
@@ -106,10 +134,22 @@ public class ReadStreamManagement {
private final RaftServer server;
private final String name;
+ private final ExecutorService requestExecutor;
ReadStreamManagement(RaftServer server) {
this.server = server;
this.name = server.getId() + "-" +
JavaUtils.getClassSimpleName(getClass());
+
+ final RaftProperties properties = server.getProperties();
+ this.requestExecutor = ConcurrentUtils.newThreadPoolWithMax(
+
RaftServerConfigKeys.DataStream.asyncRequestThreadPoolCached(properties),
+ RaftServerConfigKeys.DataStream.asyncRequestThreadPoolSize(properties),
+ name + "-request-");
+ }
+
+ void shutdown() {
+ ConcurrentUtils.shutdownAndWait(TimeDuration.ONE_SECOND, requestExecutor,
+ timeout -> LOG.warn("{}: requestExecutor shutdown timeout in {}",
this, timeout));
}
boolean process(DataStreamRequestByteBuf requestBuf, ChannelHandlerContext
ctx) {
@@ -146,8 +186,14 @@ public class ReadStreamManagement {
return true;
}
- final ReadStream stream = new ReadStream(requestBuf, ctx);
- division.getStateMachine().data().query(request.getMessage(), stream);
+ final ReadStream stream = new ReadStream(request,
requestBuf.getStreamId(), ctx);
+ requestExecutor.execute(() -> {
+ try {
+ division.getStateMachine().data().query(request.getMessage(), stream);
+ } catch (Throwable t) {
+ LOG.error("{}: Failed read-only data stream query for {}", this,
request, t);
+ }
+ });
return true;
}
diff --git
a/ratis-test/src/test/java/org/apache/ratis/netty/server/TestDataStreamManagement.java
b/ratis-test/src/test/java/org/apache/ratis/netty/server/TestDataStreamManagement.java
index 1573a2a28..516f8c171 100644
---
a/ratis-test/src/test/java/org/apache/ratis/netty/server/TestDataStreamManagement.java
+++
b/ratis-test/src/test/java/org/apache/ratis/netty/server/TestDataStreamManagement.java
@@ -124,12 +124,14 @@ class TestDataStreamManagement {
for (Object outbound; (outbound = embeddedChannel.readOutbound()) !=
null;) {
replies.add((DataStreamReply) outbound);
}
- assertEquals(1, replies.size());
+ assertEquals(2, replies.size());
}, 10, TimeDuration.valueOf(100, TimeUnit.MILLISECONDS), "read-only
replies", null);
assertEquals(query, messageRef.get().getContent());
assertFalse(streamRef.get().isOpen(), "state machine should close the
streaming query channel");
assertSuccessReply(Type.STREAM_DATA, response.size(), replies.get(0));
+ assertSuccessReply(Type.STREAM_HEADER, 0, replies.get(1));
+
assertTrue(ClientProtoUtils.getRaftClientReply(replies.get(1)).isSuccess());
} finally {
embeddedChannel.finishAndReleaseAll();
}
