amaliujia commented on code in PR #11412:
URL: https://github.com/apache/ozone/pull/11412#discussion_r4257072833
##########
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:
+1. If the Ratis handle the stream.close() and stream.fail() (which it does
already), the application won't need to worry but just either return normally
or throw proper error.
--
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]
