Copilot commented on code in PR #10369:
URL: https://github.com/apache/ozone/pull/10369#discussion_r3401553189
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java:
##########
@@ -431,14 +438,25 @@ private void readChunkDataIntoBuffers(ChunkInfo
readChunkInfo)
/**
* Send RPC call to get the chunk from the container.
+ * <p>
+ * On the zero-copy path (HDDS-10283), the returned {@link ByteBuffer}s
+ * view the underlying inbound Netty buffer of the response. Those bytes
+ * stay reachable until {@link #releaseBuffers()} releases them via
+ * {@link XceiverClientSpi#releaseReceivedResponse}, which this method
+ * registers as a side effect through {@link #lastResponseToRelease}.
*/
@VisibleForTesting
protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
throws IOException {
- ReadChunkResponseProto readChunkResponse =
- ContainerProtocolCalls.readChunk(xceiverClient, readChunkInfo,
datanodeBlockID, validators,
- tokenSupplier.get());
+ ContainerCommandResponseProto reply =
+ ContainerProtocolCalls.readChunkForZeroCopy(xceiverClient,
readChunkInfo,
+ datanodeBlockID, validators, tokenSupplier.get());
+ // The zero-copy-tracked buffer for the chunk data lives inside `reply`.
+ // Stash the outer reference so releaseBuffers() can return it once the
Review Comment:
`readChunk()` stores `reply` in `lastResponseToRelease` before validating
that the response actually contains `data`/`dataBuffers`. If the response is
malformed and the method throws, the zero-copy tracked buffer can be retained
longer than necessary. Release the response immediately on the error path (and
ideally only stash it once you know the data is usable).
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmContainerLocationCache.java:
##########
@@ -835,35 +835,49 @@ private void mockReadChunk(XceiverClientGrpc
mockDnProtocol,
byte[] data,
Exception exception,
Result errorCode) throws Exception {
- final CompletableFuture<ContainerCommandResponseProto> response;
- if (exception != null) {
- response = new CompletableFuture<>();
- response.completeExceptionally(exception);
- } else if (errorCode != null) {
- ContainerCommandResponseProto readChunkResp =
- ContainerCommandResponseProto.newBuilder()
- .setResult(errorCode)
- .setCmdType(Type.ReadChunk)
- .build();
- response = completedFuture(readChunkResp);
+ final ContainerCommandResponseProto response;
+ if (errorCode != null) {
+ response = ContainerCommandResponseProto.newBuilder()
+ .setResult(errorCode)
+ .setCmdType(Type.ReadChunk)
+ .build();
+ } else if (data != null) {
+ response = ContainerCommandResponseProto.newBuilder()
+ .setReadChunk(ReadChunkResponseProto.newBuilder()
+ .setBlockID(createBlockId(containerId, localId))
+ .setChunkData(createChunkInfo(data))
+ .setData(ByteString.copyFrom(data))
+ .build()
+ )
+ .setResult(Result.SUCCESS)
+ .setCmdType(Type.ReadChunk)
+ .build();
} else {
- ContainerCommandResponseProto readChunkResp =
- ContainerCommandResponseProto.newBuilder()
- .setReadChunk(ReadChunkResponseProto.newBuilder()
- .setBlockID(createBlockId(containerId, localId))
- .setChunkData(createChunkInfo(data))
- .setData(ByteString.copyFrom(data))
- .build()
- )
- .setResult(Result.SUCCESS)
- .setCmdType(Type.ReadChunk)
- .build();
- response = completedFuture(readChunkResp);
+ response = null;
}
Review Comment:
`response` is set to `null` when `data == null`, but the mock then passes
that `null` into validators and returns it from `sendCommandWithZeroCopy`,
which can lead to NPEs and hides test setup errors. It’s better to fail fast if
neither `data`, `errorCode`, nor `exception` is provided.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java:
##########
@@ -130,36 +140,202 @@ public XceiverClientReply sendCommandAsync(
assertEquals(0, allDNs.size());
}
- @Test
- public void testReadChunkRetryAllNodes() {
+ @ParameterizedTest(name = "readChunkApi={0}")
+ @EnumSource(ReadChunkApi.class)
+ public void testReadChunkRetryAllNodes(ReadChunkApi readChunkApi) {
final ArrayList<DatanodeDetails> allDNs = new ArrayList<>(dns);
assertThat(allDNs.size()).isGreaterThan(1);
try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf) {
@Override
- public XceiverClientReply sendCommandAsync(
+ protected XceiverClientReply sendCommandAsync(
ContainerProtos.ContainerCommandRequestProto request,
- DatanodeDetails dn) throws IOException {
+ DatanodeDetails dn, boolean zeroCopy) throws IOException {
allDNs.remove(dn);
throw new IOException("Failed " + dn);
}
}) {
- invokeXceiverClientReadChunk(client);
+ invokeXceiverClientReadChunk(client, readChunkApi);
} catch (IOException e) {
e.printStackTrace();
}
assertEquals(0, allDNs.size());
}
+ @ParameterizedTest(name = "readChunkApi={0}")
+ @EnumSource(ReadChunkApi.class)
+ public void testReadChunkApisReturnExpectedDataAndReleaseOwnership(
+ ReadChunkApi readChunkApi) throws IOException {
+ final AtomicInteger releaseCount = new AtomicInteger();
+ try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf) {
+ @Override
+ protected XceiverClientReply sendCommandAsync(
+ ContainerProtos.ContainerCommandRequestProto request,
+ DatanodeDetails dn, boolean zeroCopy) {
+ return buildValidReadChunkResponse();
+ }
+
+ @Override
+ public void releaseReceivedResponse(
+ ContainerProtos.ContainerCommandResponseProto response) {
+ releaseCount.incrementAndGet();
+ }
+ }) {
+ BlockID bid = new BlockID(1, 1);
+ bid.setBlockCommitSequenceId(1);
+ ContainerProtos.ChunkInfo chunkInfo =
+ ContainerProtos.ChunkInfo.newBuilder()
+ .setChunkName("Anything")
+ .setChecksumData(ContainerProtos.ChecksumData.newBuilder()
+ .setBytesPerChecksum(1)
+ .setType(ContainerProtos.ChecksumType.CRC32)
+ .build())
+ .setLen(1)
+ .setOffset(0)
+ .build();
+ if (readChunkApi == ReadChunkApi.ZERO_COPY) {
+ ContainerProtos.ContainerCommandResponseProto reply =
+ ContainerProtocolCalls.readChunkForZeroCopy(client, chunkInfo,
+ bid.getDatanodeBlockIDProtobuf(), null, null);
+ assertEquals(0, releaseCount.get());
+ try {
+ assertArrayEquals(new byte[] {1},
+ reply.getReadChunk().getData().toByteArray());
+ } finally {
+ client.releaseReceivedResponse(reply);
+ }
+ } else {
+ assertArrayEquals(new byte[] {1},
+ ContainerProtocolCalls.readChunk(client, chunkInfo,
+ bid.getDatanodeBlockIDProtobuf(), null, null)
+ .getData().toByteArray());
+ assertEquals(0, releaseCount.get());
+ }
+ }
+
+ assertEquals(readChunkApi == ReadChunkApi.ZERO_COPY ? 1 : 0,
+ releaseCount.get());
+ }
+
+ @Test
+ public void testReadChunkZeroCopyIsOptInOnly() throws IOException {
+ final AtomicInteger regularCallCount = new AtomicInteger();
+ final AtomicInteger zeroCopyCallCount = new AtomicInteger();
+ final BlockID blockID = new BlockID(1, 1);
+ final ContainerProtos.ChunkInfo chunkInfo = ContainerProtos.ChunkInfo
+ .newBuilder()
+ .setChunkName("Anything")
+ .setChecksumData(ContainerProtos.ChecksumData.newBuilder()
+ .setBytesPerChecksum(1)
+ .setType(ContainerProtos.ChecksumType.CRC32)
+ .build())
+ .setLen(1)
+ .setOffset(0)
+ .build();
+ final ContainerProtos.ContainerCommandRequestProto request =
+ ContainerProtos.ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.ReadChunk)
+ .setContainerID(blockID.getContainerID())
+ .setDatanodeUuid(pipeline.getFirstNode().getUuidString())
+ .setReadChunk(ContainerProtos.ReadChunkRequestProto.newBuilder()
+ .setBlockID(blockID.getDatanodeBlockIDProtobuf())
+ .setChunkData(chunkInfo))
+ .build();
+
+ try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf) {
+ @Override
+ protected XceiverClientReply sendCommandAsync(
+ ContainerProtos.ContainerCommandRequestProto request,
+ DatanodeDetails dn, boolean zeroCopy) {
+ if (zeroCopy) {
+ zeroCopyCallCount.incrementAndGet();
+ } else {
+ regularCallCount.incrementAndGet();
+ }
+ return buildValidReadChunkResponse();
+ }
+ }) {
+ assertArrayEquals(new byte[] {1},
+ client.sendCommand(request).getReadChunk().getData().toByteArray());
+ assertEquals(1, regularCallCount.get());
+ assertEquals(0, zeroCopyCallCount.get());
+
+ ContainerProtos.ContainerCommandResponseProto reply =
+ ContainerProtocolCalls.readChunkForZeroCopy(client, chunkInfo,
+ blockID.getDatanodeBlockIDProtobuf(), null, null);
+ try {
+ assertArrayEquals(new byte[] {1},
+ reply.getReadChunk().getData().toByteArray());
+ } finally {
+ client.releaseReceivedResponse(reply);
+ }
+ }
+
+ assertEquals(1, regularCallCount.get());
+ assertEquals(1, zeroCopyCallCount.get());
+ }
+
+ @Test
+ public void testReadChunkValidatorFailureReleasesResponseBeforeRetry()
+ throws IOException {
+ final ArrayList<DatanodeDetails> allDNs = new ArrayList<>(dns);
+ final AtomicInteger releaseCount = new AtomicInteger();
+ final BlockID blockID = new BlockID(1, 1);
+ final ContainerProtos.ChunkInfo chunkInfo = ContainerProtos.ChunkInfo
+ .newBuilder()
+ .setChunkName("Anything")
+ .setChecksumData(ContainerProtos.ChecksumData.newBuilder()
+ .setBytesPerChecksum(1)
+ .setType(ContainerProtos.ChecksumType.CRC32)
+ .build())
+ .setLen(1)
+ .setOffset(0)
+ .build();
+ final ContainerProtos.ContainerCommandRequestProto request =
+ ContainerProtos.ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.ReadChunk)
+ .setContainerID(blockID.getContainerID())
+ .setDatanodeUuid(pipeline.getFirstNode().getUuidString())
+ .setReadChunk(ContainerProtos.ReadChunkRequestProto.newBuilder()
+ .setBlockID(blockID.getDatanodeBlockIDProtobuf())
+ .setChunkData(chunkInfo))
+ .build();
+
+ try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf) {
+ @Override
+ protected XceiverClientReply sendCommandAsync(
+ ContainerProtos.ContainerCommandRequestProto request,
+ DatanodeDetails dn, boolean zeroCopy) {
+ allDNs.remove(dn);
+ return buildValidReadChunkResponse();
+ }
+
+ @Override
+ public void releaseReceivedResponse(
+ ContainerProtos.ContainerCommandResponseProto response) {
+ releaseCount.incrementAndGet();
+ }
+ }) {
+ IOException ex = assertThrows(IOException.class,
+ () -> client.sendCommand(request, Collections.singletonList((req,
resp) -> {
+ throw new IOException("validator failed");
+ })));
+ assertThat(ex).hasMessageContaining("validator failed");
Review Comment:
This `assertThrows` invocation exceeds the 120-character line length
enforced by the repo Checkstyle config (see `LineLength max=120`). Please wrap
the arguments across lines to avoid Checkstyle warnings.
--
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]
