HDDS-16419. Avoid nested datanode failover in GetBlock and ReadChunk

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<Validator> 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<DatanodeDetails> getDatanodesInOrder(DatanodeBlockID blockID,
+      ContainerProtos.Type cmdType) throws IOException {
+    return sortDatanodes(blockID, cmdType);
+  }
+
   private XceiverClientReply sendCommandWithTraceIDAndRetry(
       ContainerCommandRequestProto request, List<Validator> 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<DatanodeDetails> 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<Validator> 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<Validator> 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();

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(),

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<Validator> 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<DatanodeDetails> getDatanodesInOrder(
+      ContainerProtos.DatanodeBlockID blockID, ContainerProtos.Type cmdType)
+      throws IOException {
+    final Pipeline pipeline = getPipeline();
+    final List<DatanodeDetails> datanodes = new ArrayList<>(pipeline.size());
+    final Set<DatanodeDetails> 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");
   }

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> 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> T tryEachDatanode(List<DatanodeDetails> datanodes,
       CheckedFunction<DatanodeDetails, T, IOException> op,
       Function<DatanodeDetails, String> toErrorMessage)
       throws IOException {
-    final Set<DatanodeDetails> 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> 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()) {