HDDS-15521. StreamBlockInputStream fails with TimeoutIOException without retry or failover

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java

@@ -30,6 +30,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -603,9 +604,22 @@ public void streamRead(ContainerCommandRequestProto request,
 
   @Override
   public void initStreamRead(BlockID blockID, StreamingReaderSpi streamObserver) throws IOException {
+    initStreamRead(blockID, streamObserver, Collections.emptySet());
+  }
+
+  /**
+   * Start a streaming read, skipping datanodes that previously failed for this block stream.
+   */
+  public void initStreamRead(BlockID blockID, StreamingReaderSpi streamObserver,
+      Set<DatanodeID> excludedDatanodes) throws IOException {
     final List<DatanodeDetails> datanodeList = sortDatanodes(null, ContainerProtos.Type.ReadBlock);
     IOException lastException = null;
     for (DatanodeDetails dn : datanodeList) {
+      if (excludedDatanodes.contains(dn.getID())) {
+        LOG.debug("Skipping excluded datanode {} (uuid={}) for initStreamRead {}",
+            dn, dn.getUuidString(), blockID.getContainerBlockID());
+        continue;
+      }
       try {
         checkOpen(dn);
         semaphore.acquire();

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java

@@ -23,6 +23,8 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -36,6 +38,8 @@
 import org.apache.hadoop.fs.FSExceptionMessages;
 import org.apache.hadoop.hdds.StringUtils;
 import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.DatanodeID;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadBlockResponseProto;
 import org.apache.hadoop.hdds.scm.OzoneClientConfig;
@@ -47,6 +51,7 @@
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
+import org.apache.hadoop.hdds.utils.ConnectionFailureUtils;
 import org.apache.hadoop.io.retry.RetryPolicy;
 import org.apache.hadoop.ozone.common.Checksum;
 import org.apache.hadoop.ozone.common.ChecksumData;
@@ -91,6 +96,7 @@ public class StreamBlockInputStream extends BlockExtendedInputStream {
   private final Function<BlockID, BlockLocationInfo> refreshFunction;
   private final RetryPolicy retryPolicy;
   private int retries = 0;
+  private final Set<DatanodeID> failedStreamingDatanodes = new HashSet<>();
 
   public StreamBlockInputStream(
       BlockID blockID, long length, Pipeline pipeline,
@@ -171,13 +177,19 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th
     if (position >= blockLength) {
       return false;
     }
-    initialize();
-
-    if (bufferHasRemaining()) {
-      return true;
+    while (true) {
+      try {
+        initialize();
+        if (bufferHasRemaining()) {
+          return true;
+        }
+        buffer = streamingReader.read(length, preRead);
+        retries = 0;
+        return bufferHasRemaining();
+      } catch (IOException ex) {
+        handleExceptions(ex);
+      }
     }
-    buffer = streamingReader.read(length, preRead);
-    return bufferHasRemaining();
   }
 
   private synchronized void advancePosition(long delta) {
@@ -293,7 +305,7 @@ private synchronized void initialize() throws IOException {
       try {
         acquireClient();
         final StreamingReader reader = new StreamingReader();
-        xceiverClient.initStreamRead(blockID, reader);
+        xceiverClient.initStreamRead(blockID, reader, failedStreamingDatanodes);
         streamingReader = reader;
       } catch (IOException ioe) {
         handleExceptions(ioe);
@@ -327,10 +339,14 @@ synchronized void readBlockImpl(long length) throws IOException {
   }
 
   private void handleExceptions(IOException cause) throws IOException {
-    if (cause instanceof StorageContainerException || isConnectivityIssue(cause)) {
-      if (shouldRetryRead(cause, retryPolicy, retries++)) {
+    IOException root = ConnectionFailureUtils.unwrapCause(cause);
+    if (root instanceof StorageContainerException || isConnectivityIssue(root) ||
+         root instanceof TimeoutIOException) {
+      if (shouldRetryRead(root, retryPolicy, retries++)) {
+        recordFailedStreamingDatanode();
         releaseClient();
-        refreshBlockInfo(cause);
+        refreshBlockInfo(root);
+        requestedLength = position;
         LOG.warn("Refreshing block data to read block {} due to {}", blockID, cause.getMessage());
       } else {
         throw cause;
@@ -340,6 +356,21 @@ private void handleExceptions(IOException cause) throws IOException {
     }
   }
 
+  private void recordFailedStreamingDatanode() {
+    if (streamingReader == null) {
+      return;
+    }
+    final StreamingReadResponse response = streamingReader.getResponse();
+    if (response == null) {
+      return;
+    }
+    final DatanodeDetails dn = response.getDatanodeDetails();
+    if (failedStreamingDatanodes.add(dn.getID())) {
+      LOG.warn("Excluding DataNode {} from streaming read retries for block {}",
+          dn, blockID);
+    }
+  }
+
   protected synchronized void releaseClient() {
     if (xceiverClientFactory != null && xceiverClient != null) {
       closeStream();

hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java

@@ -224,7 +224,7 @@ public void testPollDoesNotDropQueuedItemWhenFutureCompletesFirst() throws Excep
       reader.setStreamingReadResponse(streamingReadResponse);
       readerRef.set(reader);
       return null;
-    }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+    }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
 
     // Simulate the race: when the client sends a ReadBlock request, the server
     // responds with data (onNext) and closes the stream (onCompleted) before
@@ -304,7 +304,7 @@ private XceiverClientGrpc mockStreamingReadClient(byte[] data,
           .setReadBlock(readBlock)
           .build());
       return null;
-    }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+    }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
 
     return xceiverClient;
   }
@@ -338,7 +338,7 @@ public void testReadDoesNotDropQueuedItemsWhenFutureIsDoneOnSecondCall() throws
       reader.setStreamingReadResponse(streamingReadResponse);
       readerRef.set(reader);
       return null;
-    }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+    }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
 
     // Server delivers both 4-byte chunks plus onCompleted() in one synchronous
     // call. After streamRead() returns: queue=[chunk1, chunk2], isDone=true.
@@ -388,7 +388,7 @@ public void testReadGetsFreshResponseTimeoutAfterStreamReadWait() throws Excepti
       reader.setStreamingReadResponse(streamingReadResponse);
       readerRef.set(reader);
       return null;
-    }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+    }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
     doAnswer(inv -> {
       Thread.sleep(450);
       Thread responseThread = new Thread(() -> {
@@ -437,7 +437,7 @@ public void testReadWithoutNewRequestGetsFreshTimeoutBudget() throws Exception {
       reader.setStreamingReadResponse(streamingReadResponse);
       readerRef.set(reader);
       return null;
-    }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+    }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
     doAnswer(inv -> {
       streamReads.incrementAndGet();
       readerRef.get().onNext(buildResponseProto(new byte[] {1}, 0));

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java

@@ -18,11 +18,15 @@
 package org.apache.hadoop.hdds.utils;
 
 import java.io.EOFException;
+import java.io.IOException;
 import java.net.ConnectException;
 import java.net.NoRouteToHostException;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.net.UnknownHostException;
+import java.util.concurrent.ExecutionException;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
+import org.apache.ratis.protocol.exceptions.TimeoutIOException;
 
 /**
  * Shared classifier for exceptions where the cached peer IP is no longer
@@ -100,4 +104,28 @@ public static boolean isConnectionFailure(Throwable t) {
     }
     return false;
   }
+
+  /**
+   * Returns the first {@link StorageContainerException} or
+   * {@link TimeoutIOException} in {@code ex}'s cause chain (through
+   * {@link ExecutionException} and nested {@link IOException} wrappers).
+   */
+  public static IOException unwrapCause(IOException ex) {
+    Throwable t = ex;
+    while (t != null) {
+      if (t instanceof TimeoutIOException || t instanceof StorageContainerException) {
+        return (IOException) t;
+      }
+      if (t instanceof ExecutionException && t.getCause() != null) {
+        t = t.getCause();
+        continue;
+      }
+      if (t.getCause() instanceof IOException) {
+        t = t.getCause();
+        continue;
+      }
+      break;
+    }
+    return ex;
+  }
 }