From 24cfab9fa9068d5f93f5b809d1ac97d8d2005624 Mon Sep 17 00:00:00 2001 From: Symious Date: Tue, 15 Sep 2026 10:16:43 +0800 Subject: [PATCH] HDDS-16419. Avoid nested datanode failover in GetBlock and ReadChunk Co-Authored-By: Claude Opus 5 --- .../hadoop/hdds/scm/XceiverClientGrpc.java | 120 +++++++++++++----- .../scm/storage/TestChunkInputStream.java | 4 +- .../hadoop/hdds/scm/XceiverClientSpi.java | 42 ++++++ .../scm/storage/ContainerProtocolCalls.java | 30 +++-- .../hdds/scm/TestXceiverClientGrpc.java | 88 +++++++++---- 5 files changed, 214 insertions(+), 70 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java index 93e7c0059340..829b6293dcb9 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java @@ -407,22 +407,57 @@ public ContainerCommandResponseProto sendCommand( } } + @Override + public ContainerCommandResponseProto sendCommand( + ContainerCommandRequestProto request, List validators, + DatanodeDetails datanode) throws IOException { + return TracingUtil.executeInNewSpan(getSpanName(request), SpanKind.CLIENT, + () -> { + final ContainerCommandRequestProto requestWithTraceID = withTraceIDAndVersion(request); + try { + return sendCommandToDatanode(requestWithTraceID, validators, datanode); + } catch (ExecutionException e) { + if (LOG.isDebugEnabled()) { + LOG.debug("Failed to execute command {} on datanode {}", + processForDebug(requestWithTraceID), datanode, e); + } + throw toIOException(e); + } catch (InterruptedException e) { + LOG.error("Command execution was interrupted ", e); + Thread.currentThread().interrupt(); + throw (IOException) new InterruptedIOException( + "Command " + processForDebug(requestWithTraceID) + " was interrupted.") + .initCause(e); + } + }); + } + + @Override + public List getDatanodesInOrder(DatanodeBlockID blockID, + ContainerProtos.Type cmdType) throws IOException { + return sortDatanodes(blockID, cmdType); + } + private XceiverClientReply sendCommandWithTraceIDAndRetry( ContainerCommandRequestProto request, List validators) throws IOException { + return TracingUtil.executeInNewSpan(getSpanName(request), SpanKind.CLIENT, + () -> sendCommandWithRetry(withTraceIDAndVersion(request), validators)); + } - String spanName = "XceiverClientGrpc." + request.getCmdType().name(); + private static String getSpanName(ContainerCommandRequestProto request) { + return "XceiverClientGrpc." + request.getCmdType().name(); + } - return TracingUtil.executeInNewSpan(spanName, SpanKind.CLIENT, - () -> { - ContainerCommandRequestProto.Builder builder = - ContainerCommandRequestProto.newBuilder(request) - .setTraceID(TracingUtil.exportCurrentSpan()); - if (!request.hasVersion()) { - builder.setVersion(ClientVersion.CURRENT.toProtoValue()); - } - return sendCommandWithRetry(builder.build(), validators); - }); + private static ContainerCommandRequestProto withTraceIDAndVersion( + ContainerCommandRequestProto request) { + ContainerCommandRequestProto.Builder builder = + ContainerCommandRequestProto.newBuilder(request) + .setTraceID(TracingUtil.exportCurrentSpan()); + if (!request.hasVersion()) { + builder.setVersion(ClientVersion.CURRENT.toProtoValue()); + } + return builder.build(); } private List sortDatanodes(ContainerCommandRequestProto request) throws IOException { @@ -485,6 +520,45 @@ private static DatanodeBlockID getRequestBlockID(ContainerCommandRequestProto re return blockID; } + /** + * Sends the command to the given datanode only and validates the response. + */ + private ContainerCommandResponseProto sendCommandToDatanode( + ContainerCommandRequestProto request, List validators, + DatanodeDetails dn) + throws IOException, ExecutionException, InterruptedException { + if (LOG.isDebugEnabled()) { + LOG.debug("Executing command {} on datanode {}", + processForDebug(request), dn); + } + final ContainerCommandResponseProto responseProto = + sendCommandAsync(request, dn).getResponse().get(); + if (validators != null && !validators.isEmpty()) { + for (Validator validator : validators) { + validator.accept(request, responseProto); + } + } + if (request.getCmdType() == ContainerProtos.Type.GetBlock) { + DatanodeBlockID getBlockID = request.getGetBlock().getBlockID(); + getBlockDNcache.put(getBlockID, dn); + } + return responseProto; + } + + /** + * @return the IOException for a failed command + * @throws SCMSecurityException if the datanode rejected the block token + */ + private static IOException toIOException(ExecutionException e) + throws SCMSecurityException { + if (Status.fromThrowable(e.getCause()).getCode() + == Status.UNAUTHENTICATED.getCode()) { + throw new SCMSecurityException("Failed to authenticate with " + + "GRPC XceiverServer with Ozone block token."); + } + return new IOException(e); + } + private XceiverClientReply sendCommandWithRetry( ContainerCommandRequestProto request, List validators) throws IOException { @@ -498,28 +572,14 @@ private XceiverClientReply sendCommandWithRetry( for (DatanodeDetails dn : datanodeList) { try { - if (LOG.isDebugEnabled()) { - LOG.debug("Executing command {} on datanode {}", - processForDebug(request), dn); - } // In case the command gets retried on a 2nd datanode, // sendCommandAsyncCall will create a new channel and async stub // in case these don't exist for the specific datanode. reply.addDatanode(dn); - responseProto = sendCommandAsync(request, dn).getResponse().get(); - if (validators != null && !validators.isEmpty()) { - for (Validator validator : validators) { - validator.accept(request, responseProto); - } - } - if (request.getCmdType() == ContainerProtos.Type.GetBlock) { - DatanodeBlockID getBlockID = request.getGetBlock().getBlockID(); - getBlockDNcache.put(getBlockID, dn); - } + responseProto = sendCommandToDatanode(request, validators, dn); break; } catch (IOException e) { ioException = e; - responseProto = null; if (LOG.isDebugEnabled()) { LOG.debug("Failed to execute command {} on datanode {}", processForDebug(request), dn, e); @@ -529,13 +589,7 @@ private XceiverClientReply sendCommandWithRetry( LOG.debug("Failed to execute command {} on datanode {}", processForDebug(request), dn, e); } - if (Status.fromThrowable(e.getCause()).getCode() - == Status.UNAUTHENTICATED.getCode()) { - throw new SCMSecurityException("Failed to authenticate with " - + "GRPC XceiverServer with Ozone block token."); - } - - ioException = new IOException(e); + ioException = toIOException(e); } catch (InterruptedException e) { LOG.error("Command execution was interrupted ", e); Thread.currentThread().interrupt(); diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java index 248ea8655223..49b8595b028b 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java @@ -261,7 +261,9 @@ public void connectsToNewPipeline() throws Exception { ArgumentCaptor.forClass(ContainerCommandRequestProto.class); when(client.getPipeline()) .thenAnswer(invocation -> pipelineRef.get()); - when(client.sendCommand(requestCaptor.capture(), any())) + when(client.getDatanodesInOrder(any(), any())) + .thenAnswer(invocation -> pipelineRef.get().getNodes()); + when(client.sendCommand(requestCaptor.capture(), any(), any())) .thenAnswer(invocation -> getReadChunkResponse( requestCaptor.getValue(), diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java index 60d3a1d7b70a..6451bd54da5a 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientSpi.java @@ -21,8 +21,11 @@ import java.io.Closeable; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicInteger; @@ -159,6 +162,45 @@ public ContainerCommandResponseProto sendCommand( } } + /** + * Sends a given command to the given datanode only, without failing over + * to the other datanodes in the pipeline. + * Implementations which cannot target a specific datanode fall back to + * {@link #sendCommand(ContainerCommandRequestProto, List)}. + * @param request Request + * @param validators functions to validate the response + * @param datanode the datanode to send the command to + * @return Response to the command + */ + public ContainerCommandResponseProto sendCommand( + ContainerCommandRequestProto request, + List validators, + DatanodeDetails datanode) + throws IOException { + return sendCommand(request, validators); + } + + /** + * Returns the datanodes of the pipeline in the order they should be tried + * for a command on the given block. + * @param blockID the block the command operates on + * @param cmdType type of the command + * @return datanodes of the pipeline, in the order to try + */ + public List getDatanodesInOrder( + ContainerProtos.DatanodeBlockID blockID, ContainerProtos.Type cmdType) + throws IOException { + final Pipeline pipeline = getPipeline(); + final List datanodes = new ArrayList<>(pipeline.size()); + final Set excluded = new HashSet<>(); + while (excluded.size() < pipeline.size()) { + final DatanodeDetails d = pipeline.getClosestNode(excluded); + datanodes.add(d); + excluded.add(d); + } + return datanodes; + } + public void initStreamRead(BlockID blockID, StreamingReaderSpi streamObserver) throws IOException { throw new UnsupportedOperationException("Stream read is not supported"); } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java index b8d832cf45a4..d747d7b68614 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java @@ -25,10 +25,8 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; -import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.function.Function; import org.apache.hadoop.hdds.annotation.InterfaceStability; @@ -143,13 +141,20 @@ public static ListBlockResponseProto listBlock(XceiverClientSpi xceiverClient, return response.getListBlock(); } - static T tryEachDatanode(Pipeline pipeline, + /** + * Applies {@code op} to each datanode in order until it succeeds. + * {@code op} must only contact the given datanode, otherwise each attempt + * would fail over to the whole pipeline again. + */ + static T tryEachDatanode(List datanodes, CheckedFunction op, Function toErrorMessage) throws IOException { - final Set excluded = new HashSet<>(); - for (; ;) { - final DatanodeDetails d = pipeline.getClosestNode(excluded); + if (datanodes.isEmpty()) { + throw new IOException("No datanode to try"); + } + for (int i = 0; ; i++) { + final DatanodeDetails d = datanodes.get(i); try { return op.apply(d); @@ -165,8 +170,7 @@ static T tryEachDatanode(Pipeline pipeline, } } span.addEvent("failed to connect to DN " + d); - excluded.add(d); - if (excluded.size() < pipeline.size()) { + if (i < datanodes.size() - 1) { LOG.warn(toErrorMessage.apply(d) + "; will try another datanode.", e); } else { @@ -197,7 +201,8 @@ public static GetBlockResponseProto getBlock(XceiverClientSpi xceiverClient, builder.setEncodedToken(token.encodeToUrlString()); } - return tryEachDatanode(xceiverClient.getPipeline(), + return tryEachDatanode( + xceiverClient.getDatanodesInOrder(blockID.getDatanodeBlockIDProtobuf(), Type.GetBlock), d -> getBlock(xceiverClient, validators, builder, blockID, d, pipeline), d -> toErrorMessage(blockID, d)); } @@ -256,7 +261,7 @@ private static GetBlockResponseProto getBlock(XceiverClientSpi xceiverClient, .setDatanodeUuid(datanode.getUuidString()) .setGetBlock(readBlockRequest).build(); ContainerCommandResponseProto response = - xceiverClient.sendCommand(request, validators); + xceiverClient.sendCommand(request, validators, datanode); return response.getGetBlock(); } @@ -431,7 +436,8 @@ public static ContainerProtos.ReadChunkResponseProto readChunk( span.setAttribute("offset", chunk.getOffset()) .setAttribute("length", chunk.getLen()) .setAttribute("block", blockID.toString()); - return tryEachDatanode(xceiverClient.getPipeline(), + return tryEachDatanode( + xceiverClient.getDatanodesInOrder(blockID, Type.ReadChunk), d -> readChunk(xceiverClient, chunk, blockID, validators, builder, d), d -> toErrorMessage(chunk, blockID, d)); @@ -450,7 +456,7 @@ private static ContainerProtos.ReadChunkResponseProto readChunk( requestBuilder = requestBuilder.setTraceID(traceId); } ContainerCommandResponseProto reply = - xceiverClient.sendCommand(requestBuilder.build(), validators); + xceiverClient.sendCommand(requestBuilder.build(), validators, d); final ReadChunkResponseProto response = reply.getReadChunk(); final long readLen = getLen(response); if (readLen != chunk.getLen()) { diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java index 8cfcac043343..4d80a13f49e2 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java @@ -45,6 +45,7 @@ import org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls; import org.apache.hadoop.ozone.OzoneConfigKeys; import org.apache.ratis.protocol.exceptions.TimeoutIOException; +import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -114,43 +115,62 @@ public XceiverClientReply sendCommandAsync( } @Test - public void testGetBlockRetryAlNodes() { - final ArrayList allDNs = new ArrayList<>(dns); - assertThat(allDNs.size()).isGreaterThan(1); + public void testGetBlockRetryAlNodes() throws IOException { + final List seenDNs = new ArrayList<>(); + try (XceiverClientGrpc client = failingClient(seenDNs)) { + assertThrows(IOException.class, () -> invokeXceiverClientGetBlock(client)); + } + // Each datanode is tried exactly once, not once per datanode tried. + assertThat(seenDNs).containsExactlyInAnyOrderElementsOf(dns); + } + + @Test + public void testReadChunkRetryAllNodes() throws IOException { + final List seenDNs = new ArrayList<>(); + try (XceiverClientGrpc client = failingClient(seenDNs)) { + assertThrows(IOException.class, () -> invokeXceiverClientReadChunk(client)); + } + // Each datanode is tried exactly once, not once per datanode tried. + assertThat(seenDNs).containsExactlyInAnyOrderElementsOf(dns); + } + + @Test + public void testReadChunkRetryNextNodeOnShortRead() throws IOException { + final int chunkLen = 10; + final List seenDNs = new ArrayList<>(); try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf) { @Override public XceiverClientReply sendCommandAsync( ContainerProtos.ContainerCommandRequestProto request, - DatanodeDetails dn) throws IOException { - allDNs.remove(dn); - throw new IOException("Failed " + dn); + DatanodeDetails dn) { + seenDNs.add(dn); + // The first datanode returns less data than requested. + return buildReadChunkResponse(request, seenDNs.size() == 1 ? chunkLen - 1 : chunkLen); } }) { - invokeXceiverClientGetBlock(client); - } catch (IOException e) { - e.printStackTrace(); + ContainerProtos.ReadChunkResponseProto response = + invokeXceiverClientReadChunk(client, chunkLen); + assertEquals(chunkLen, response.getData().size()); } - assertEquals(0, allDNs.size()); + assertEquals(2, seenDNs.size()); + assertNotEquals(seenDNs.get(0), seenDNs.get(1)); } - @Test - public void testReadChunkRetryAllNodes() { - final ArrayList allDNs = new ArrayList<>(dns); - assertThat(allDNs.size()).isGreaterThan(1); - try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf) { + /** + * @return a client failing all commands, recording the target datanodes + */ + private XceiverClientGrpc failingClient(List seenDNs) { + return new XceiverClientGrpc(pipeline, conf) { @Override public XceiverClientReply sendCommandAsync( ContainerProtos.ContainerCommandRequestProto request, DatanodeDetails dn) throws IOException { - allDNs.remove(dn); + seenDNs.add(dn); + // The request must be addressed to the datanode it is sent to. + assertEquals(dn.getUuidString(), request.getDatanodeUuid()); throw new IOException("Failed " + dn); } - }) { - invokeXceiverClientReadChunk(client); - } catch (IOException e) { - e.printStackTrace(); - } - assertEquals(0, allDNs.size()); + }; } @Test @@ -274,22 +294,42 @@ private void invokeXceiverClientGetBlock(XceiverClientSpi client) private void invokeXceiverClientReadChunk(XceiverClientSpi client) throws IOException { + invokeXceiverClientReadChunk(client, -1); + } + + private ContainerProtos.ReadChunkResponseProto invokeXceiverClientReadChunk( + XceiverClientSpi client, long len) throws IOException { BlockID bid = new BlockID(1, 1); bid.setBlockCommitSequenceId(1); - ContainerProtocolCalls.readChunk(client, + return ContainerProtocolCalls.readChunk(client, ContainerProtos.ChunkInfo.newBuilder() .setChunkName("Anything") .setChecksumData(ContainerProtos.ChecksumData.newBuilder() .setBytesPerChecksum(512) .setType(ContainerProtos.ChecksumType.CRC32) .build()) - .setLen(-1) + .setLen(len) .setOffset(0) .build(), bid.getDatanodeBlockIDProtobuf(), null, null); } + private XceiverClientReply buildReadChunkResponse( + ContainerProtos.ContainerCommandRequestProto request, int len) { + ContainerProtos.ReadChunkRequestProto readChunk = request.getReadChunk(); + ContainerProtos.ContainerCommandResponseProto resp = + ContainerProtos.ContainerCommandResponseProto.newBuilder() + .setCmdType(ContainerProtos.Type.ReadChunk) + .setResult(ContainerProtos.Result.SUCCESS) + .setReadChunk(ContainerProtos.ReadChunkResponseProto.newBuilder() + .setBlockID(readChunk.getBlockID()) + .setChunkData(readChunk.getChunkData()) + .setData(ByteString.copyFrom(new byte[len]))) + .build(); + return new XceiverClientReply(CompletableFuture.completedFuture(resp)); + } + private void invokeXceiverClientReadSmallFile(XceiverClientSpi client) throws IOException { BlockID bid = new BlockID(1, 1);