amaliujia commented on code in PR #11412:
URL: https://github.com/apache/ozone/pull/11412#discussion_r4257117206
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java:
##########
@@ -892,6 +901,97 @@ public CompletableFuture<Message> query(Message request) {
}
}
+ /** Serves a Ratis read-only data stream. Only ReadBlock is supported, see
{@link #streamReadBlock}. */
+ @Override
+ public long transferTo(Message request, WritableByteChannel stream) throws
IOException {
+ try {
+ metrics.incNumQueryStateMachineOps();
+ final ContainerCommandRequestProto requestProto =
message2ContainerCommandRequestProto(request);
+ if (requestProto.getCmdType() != Type.ReadBlock) {
+ throw new IOException("Unsupported command for a read-only data
stream: " + requestProto.getCmdType());
+ }
+ return streamReadBlock(dispatcher, requestProto, stream);
+ } catch (IOException e) {
+ metrics.incNumQueryStateMachineFails();
+ throw e;
+ }
+ }
+
+ /**
+ * Reads a block and writes each ReadBlock response to the stream as one
reply:
+ * {@code [int: metadata length][metadata: the response without its
data][data]}.
+ * Then closes the stream, which sends the terminal reply.
+ * If the read fails, this method throws without closing the stream. Ratis
sends the failure as the terminal reply.
+ *
+ * @return the number of bytes written to the stream
+ */
+ static long streamReadBlock(ContainerDispatcher dispatcher,
ContainerCommandRequestProto request,
+ WritableByteChannel stream) throws IOException {
+ final AtomicLong bytesWritten = new AtomicLong();
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final StreamObserver<ContainerCommandResponseProto> observer = new
StreamObserver<ContainerCommandResponseProto>() {
+ @Override
+ public void onNext(ContainerCommandResponseProto response) {
+ try {
+ bytesWritten.addAndGet(writeReadBlockReply(stream, response));
+ } catch (IOException e) {
+ error.set(e);
+ throw Status.CANCELLED.withDescription("Failed to write a ReadBlock
reply").withCause(e).asRuntimeException();
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ error.compareAndSet(null, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ };
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ dispatcher.streamDataReadOnly(request, observer, blockFile, null);
+ } catch (StatusRuntimeException e) {
+ error.compareAndSet(null, e);
+ }
+ if (error.get() != null) {
+ throw new IOException("Failed to read block " +
BlockID.getFromProtobuf(request.getReadBlock().getBlockID()),
+ error.get());
+ }
+ stream.close();
Review Comment:
Going back to now without the proper stream.close() on the Ratis side, if we
check the existing Ratis implementation
https://github.com/apache/ratis/blob/5396d171f4d1a5513b6713d06362c6ef56ba1beb/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java#L252-L265
The minimal is to ensure that the stream only closes when return normally,
otherwise Ratis won't be able to send back a proper `DataStreamException` to
the client.
From this perspective, the current `stream.close()` in this
`streamReadBlock` seems to ok.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
