Copilot commented on code in PR #10743:
URL: https://github.com/apache/ozone/pull/10743#discussion_r3573417086
##########
pom.xml:
##########
@@ -188,7 +188,7 @@
<ratis-thirdparty.netty.version>4.1.130.Final</ratis-thirdparty.netty.version>
<ratis-thirdparty.protobuf.version>3.25.8</ratis-thirdparty.protobuf.version>
<ratis.thirdparty.version>1.0.11</ratis.thirdparty.version>
- <ratis.version>3.2.1</ratis.version>
+ <ratis.version>3.3.0-SNAPSHOT</ratis.version>
Review Comment:
Setting the project-wide <ratis.version> to a -SNAPSHOT makes builds
non-reproducible and will fail in CI unless the snapshot repository is
configured everywhere. This should be isolated behind a Maven profile (or
similar opt-in mechanism) so the default build continues to use a released
Ratis version.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java:
##########
@@ -89,7 +92,15 @@ public BlockExtendedInputStream create(ReplicationConfig
repConfig,
return new ECBlockInputStreamProxy((ECReplicationConfig)repConfig,
blockInfo, xceiverFactory, refreshFunction,
ecBlockStreamFactory, config);
- } else if (config.isStreamReadBlock() &&
allDataNodesSupportStreamBlock(pipeline)) {
+ } else if (config.isRatisStreamReadBlock()
+ &&
repConfig.getReplicationType().equals(HddsProtos.ReplicationType.RATIS)
+ && allDataNodesSupport(pipeline,
+ RATIS_DATASTREAM_READ_BLOCK_SUPPORT)) {
+ return new RatisDataStreamBlockInputStream(blockInfo.getBlockID(),
+ blockInfo.getLength(), pipeline, token, xceiverFactory,
+ config);
Review Comment:
The Ratis data-stream read path is selected solely by datanode version, but
the RATIS_DATASTREAM port is only published when datastream is enabled on the
datanode. If the client enables ratis stream reads while datastream is disabled
server-side, create() will pick RatisDataStreamBlockInputStream and fail when
the constructor validates the missing port. Add a port-capability check (or
fallback) before selecting this stream implementation.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java:
##########
@@ -846,6 +860,222 @@ public CompletableFuture<Message> query(Message request) {
}
}
+ @Override
+ public void query(Message request, WritableByteChannel stream) {
+ try {
+ metrics.incNumQueryStateMachineOps();
+ final ContainerCommandRequestProto requestProto =
+ message2ContainerCommandRequestProto(request);
+ if (requestProto.getCmdType() == Type.ReadBlock) {
+ streamReadBlock(dispatcher, requestProto, stream);
+ } else {
+ writeAndClose(stream, dispatchCommand(requestProto, null));
+ }
+ } catch (IOException e) {
+ metrics.incNumQueryStateMachineFails();
+ throw new CompletionException(e);
+ }
+ }
+
+ static void streamReadBlock(
+ ContainerDispatcher dispatcher,
+ ContainerCommandRequestProto requestProto,
+ WritableByteChannel stream) throws IOException {
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final AtomicBoolean responseSeen = new AtomicBoolean(false);
+ final DataStreamObserver<ReadBlockResponse> observer =
+ new DataStreamObserver<ReadBlockResponse>() {
+ @Override
+ public void onNext(ReadBlockResponse response) {
+ responseSeen.set(true);
+ try {
+ writeReadBlockStreamResponse(stream, response.getResponse(),
+ response.getData());
+ } catch (IOException e) {
+ error.compareAndSet(null, e);
+ throw new CompletionException(e);
+ } catch (RuntimeException e) {
+ error.compareAndSet(null, e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ error.set(throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+
+ };
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null);
+ } catch (CompletionException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ }
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ e.getCause() != null ? e.getCause() : e);
+ }
+
+ if (error.get() != null) {
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ error.get());
+ }
+ if (!responseSeen.get()) {
+ throw new IOException("ReadBlock stream returned no responses: "
+ + requestProto);
+ }
+
+ stream.close();
+ }
Review Comment:
streamReadBlock only closes the WritableByteChannel on the success path. If
dispatching or validation throws, the channel is left open, which can leak
resources and/or leave the client-side stream hanging. Close the channel in a
finally block (and avoid overriding the primary failure with a close failure).
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java:
##########
@@ -846,6 +860,222 @@ public CompletableFuture<Message> query(Message request) {
}
}
+ @Override
+ public void query(Message request, WritableByteChannel stream) {
+ try {
+ metrics.incNumQueryStateMachineOps();
+ final ContainerCommandRequestProto requestProto =
+ message2ContainerCommandRequestProto(request);
+ if (requestProto.getCmdType() == Type.ReadBlock) {
+ streamReadBlock(dispatcher, requestProto, stream);
+ } else {
+ writeAndClose(stream, dispatchCommand(requestProto, null));
+ }
+ } catch (IOException e) {
+ metrics.incNumQueryStateMachineFails();
+ throw new CompletionException(e);
+ }
+ }
+
+ static void streamReadBlock(
+ ContainerDispatcher dispatcher,
+ ContainerCommandRequestProto requestProto,
+ WritableByteChannel stream) throws IOException {
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final AtomicBoolean responseSeen = new AtomicBoolean(false);
+ final DataStreamObserver<ReadBlockResponse> observer =
+ new DataStreamObserver<ReadBlockResponse>() {
+ @Override
+ public void onNext(ReadBlockResponse response) {
+ responseSeen.set(true);
+ try {
+ writeReadBlockStreamResponse(stream, response.getResponse(),
+ response.getData());
+ } catch (IOException e) {
+ error.compareAndSet(null, e);
+ throw new CompletionException(e);
+ } catch (RuntimeException e) {
+ error.compareAndSet(null, e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ error.set(throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+
+ };
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null);
+ } catch (CompletionException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ }
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ e.getCause() != null ? e.getCause() : e);
+ }
+
+ if (error.get() != null) {
+ throw new IOException("Failed to stream ReadBlock " + requestProto,
+ error.get());
+ }
+ if (!responseSeen.get()) {
+ throw new IOException("ReadBlock stream returned no responses: "
+ + requestProto);
+ }
+
+ stream.close();
+ }
+
+ private static void writeAndClose(
+ WritableByteChannel stream, ContainerCommandResponseProto response)
+ throws IOException {
+ try {
+ writeFully(stream, response.toByteString().asReadOnlyByteBuffer());
+ } finally {
+ stream.close();
+ }
+ }
+
+ private static void writeReadBlockStreamResponse(
+ WritableByteChannel stream, ContainerCommandResponseProto response,
+ ByteBuffer data) throws IOException {
+ writeFully(stream, encodeReadBlockStreamResponse(response, data));
+ }
+
+ private static void writeFully(WritableByteChannel stream, ByteBuffer buffer)
+ throws IOException {
+ while (buffer.hasRemaining()) {
+ final int position = buffer.position();
+ final int remaining = buffer.remaining();
+ final int written = stream.write(buffer);
+ if (written < 0) {
+ throw new IOException("EOF reached while writing ReadBlock stream
response");
+ }
+ if (written == 0) {
+ throw new IOException("Zero bytes written to ReadBlock stream response
channel");
+ }
+ final int advanced = buffer.position() - position;
+ if (advanced == 0) {
+ if (written > remaining) {
+ throw new IOException("ReadBlock stream response write exceeded
remaining bytes: written="
+ + written + ", remaining=" + remaining);
+ }
+ buffer.position(position + written);
+ } else if (advanced != written) {
+ throw new IOException("Inconsistent ReadBlock stream response write:
written="
+ + written + ", advanced=" + advanced);
+ }
+ }
+ }
+
+ private static ByteBuffer encodeReadBlockStreamResponse(
+ ContainerCommandResponseProto response, ByteBuffer dataBuffer) {
+ final ByteBuffer data;
+ final ContainerCommandResponseProto metadata;
+ if (response.hasReadBlock()) {
+ final ReadBlockResponseProto readBlock = response.getReadBlock();
+ data = dataBuffer != null ? dataBuffer.asReadOnlyBuffer()
+ : readBlock.getData().asReadOnlyByteBuffer();
+ metadata = response.toBuilder()
+ .setReadBlock(readBlock.toBuilder().setData(ByteString.EMPTY))
+ .build();
+ } else {
+ data = ByteBuffer.allocate(0);
+ metadata = response;
+ }
+
+ final ByteBuffer metadataBuffer =
+ metadata.toByteString().asReadOnlyByteBuffer();
+ final ByteBuffer header =
+ ByteBuffer.allocate(RATIS_READ_BLOCK_STREAM_HEADER_BYTES);
+ header.putInt(metadataBuffer.remaining());
+ header.flip();
+ final ByteBuffer frame = ByteBuffer.allocate(
+ header.remaining() + metadataBuffer.remaining() + data.remaining());
+ frame.put(header);
+ frame.put(metadataBuffer);
+ frame.put(data);
+ frame.flip();
+ return frame.asReadOnlyBuffer();
+ }
+
+ private ContainerCommandResponseProto queryReadBlock(
+ ContainerCommandRequestProto requestProto) throws IOException {
+ final List<ContainerCommandResponseProto> responses = new ArrayList<>();
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+ final DataStreamObserver<ReadBlockResponse> observer =
+ new DataStreamObserver<ReadBlockResponse>() {
+ @Override
+ public void onNext(ReadBlockResponse response) {
+ responses.add(response.getData() == null ? response.getResponse()
+ : response.getResponse().toBuilder()
+
.setReadBlock(response.getResponse().getReadBlock().toBuilder()
+ .setData(ByteString.copyFrom(response.getData())))
+ .build());
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ error.set(throwable);
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ };
+
+ try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+ dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null);
+ }
+
+ if (error.get() != null) {
+ throw new IOException("Failed to query ReadBlock " + requestProto,
+ error.get());
+ }
+ if (responses.isEmpty()) {
+ throw new IOException("ReadBlock query returned no responses: "
+ + requestProto);
+ }
+ return mergeReadBlockResponses(responses);
+ }
+
+ private static ContainerCommandResponseProto mergeReadBlockResponses(
+ List<ContainerCommandResponseProto> responses) {
+ final ContainerCommandResponseProto first = responses.get(0);
+ if (responses.size() == 1
+ || first.getResult() != ContainerProtos.Result.SUCCESS
+ || !first.hasReadBlock()) {
+ return first;
+ }
+
+ ByteString data = ByteString.EMPTY;
+ final ReadBlockResponseProto firstReadBlock = first.getReadBlock();
+ final ContainerProtos.ChecksumData.Builder checksum =
+ firstReadBlock.getChecksumData().toBuilder().clearChecksums();
+ for (ContainerCommandResponseProto response : responses) {
+ if (response.getResult() != ContainerProtos.Result.SUCCESS
+ || !response.hasReadBlock()) {
+ return response;
+ }
+ final ReadBlockResponseProto readBlock = response.getReadBlock();
+ data = data.concat(readBlock.getData());
+ checksum.addAllChecksums(readBlock.getChecksumData().getChecksumsList());
+ }
+
Review Comment:
mergeReadBlockResponses concatenates ByteString in a loop (data =
data.concat(...)), which can create excessive intermediate objects and degrade
significantly for large reads split across many responses. Accumulate into a
single output buffer instead.
--
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]
