HADOOP-19105. Improve resilience in vector reads.

Rigorous aborting of streams where appropriate, and
if one range read fails in the fallback sequential read,
so do all the others.

Change-Id: Ifc5d0c40b369f2116137a6b0ecad7a242ec044cc

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/VectoredReadUtils.java

@@ -196,6 +196,8 @@ public static CompletableFuture<ByteBuffer> readRangeFrom(
    * @param range file range
    * @param buffer destination buffer
    * @throws IOException IO problems.
+   * @throws EOFException the end of the data was reached before
+   * the read operation completed
    */
   private static void readNonByteBufferPositionedReadable(
       PositionedReadable stream,

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInputStream.java

@@ -77,6 +77,8 @@
 import static org.apache.hadoop.fs.VectoredReadUtils.mergeSortedRanges;
 import static org.apache.hadoop.fs.VectoredReadUtils.validateAndSortRanges;
 import static org.apache.hadoop.fs.s3a.Invoker.onceTrackingDuration;
+import static org.apache.hadoop.fs.s3a.impl.ErrorTranslation.shouldInputStreamBeAborted;
+import static org.apache.hadoop.fs.s3a.impl.SDKStreamDrainer.abortSdkStream;
 import static org.apache.hadoop.util.StringUtils.toLowerCase;
 import static org.apache.hadoop.util.functional.FutureIO.awaitFuture;
 
@@ -571,11 +573,8 @@ public synchronized int read() throws IOException {
           }
           try {
             b = wrappedStream.read();
-          } catch (HttpChannelEOFException | SocketTimeoutException e) {
-            onReadFailure(e, true);
-            throw e;
           } catch (IOException e) {
-            onReadFailure(e, false);
+            onReadFailure(e, shouldInputStreamBeAborted(e));
             throw e;
           }
           return b;
@@ -1092,9 +1091,18 @@ private void readCombinedRangeAndUpdateChildren(CombinedFileRange combinedFileRa
 
       // Attempt to recover from the failure by reading each range individually
       // within the current thread.
-      // If a single read is unrecoverable, all subsequent range reads are skipped.
+      // If a single read is unrecoverable, all subsequent range reads are failed
+      // with the same exception.
+      // this is to process unrecoverable failures faster.
+      IOException lastIOE = null;
       for (FileRange child : unreadRanges) {
-        readSingleRangeWithRetries(child, bufferPool);
+        if (lastIOE == null) {
+          // all good so far: request the next range
+          lastIOE = readSingleRangeWithRetries(child, bufferPool);
+        } else {
+          // a predecessor failed, do not attempt to recover.
+          child.getData().completeExceptionally(lastIOE);
+        }
       }
     } finally {
       IOUtils.cleanupWithLogger(LOG, rangeContent);
@@ -1116,7 +1124,7 @@ private void readCombinedRangeAndUpdateChildren(CombinedFileRange combinedFileRa
    */
   @Retries.OnceTranslated
   private void populateChildBuffers(CombinedFileRange combinedFileRange,
-      InputStream objectContent,
+      ResponseInputStream<GetObjectResponse> objectContent,
       ByteBufferPool bufferPool) throws IOException {
     // If the combined file range just contains a single child
     // range, we only have to fill that one child buffer else
@@ -1146,6 +1154,7 @@ private void populateChildBuffers(CombinedFileRange combinedFileRange,
             // work out how much
             long drainQuantity = child.getOffset() - position;
             // and drain it.
+            // this will raise EOFException if a -1 was returned.
             drainUnnecessaryData(objectContent, position, drainQuantity);
           }
         }
@@ -1155,6 +1164,7 @@ private void populateChildBuffers(CombinedFileRange combinedFileRange,
         } catch (IOException e) {
           // release the buffer
           bufferPool.putBuffer(buffer);
+          // rethrow
           throw e;
         }
         child.getData().complete(buffer);
@@ -1174,7 +1184,7 @@ private void populateChildBuffers(CombinedFileRange combinedFileRange,
    */
   @Retries.OnceTranslated
   private void drainUnnecessaryData(
-      final InputStream objectContent,
+      final ResponseInputStream<GetObjectResponse> objectContent,
       final long position,
       long drainQuantity) throws IOException {
 
@@ -1198,38 +1208,69 @@ private void drainUnnecessaryData(
               "End of stream reached draining data between ranges; expected %,d bytes;"
                   + " only drained %,d bytes before -1 returned (position=%,d)",
               drainQuantity, drainBytes, position + drainBytes);
+          LOG.debug(s);
           throw new EOFException(s);
         }
         drainBytes += readCount;
         remaining -= readCount;
       }
+    } catch(IOException ex) {
+      if (shouldInputStreamBeAborted(ex)) {
+        // abort the stream if the exception indicates this is needed.
+        abortSdkStream(uri, objectContent, streamStatistics, "drain failure");
+      }
+      throw ex;
     } finally {
       streamStatistics.readVectoredBytesDiscarded(drainBytes);
       LOG.debug("{} bytes drained from stream ", drainBytes);
     }
   }
 
-
   /**
    * Read data from S3 for this range and populate a buffer.
+   * The GET request and single range reads are retried.
+   * Any IOException which is propagated by the retry logic is
+   * attached to the range as an exceptional failure.
    * @param range range of data to read.
    * @param bufferPool buffer allocator.
+   * @return any IOE which resulted in the read being unsuccessful; null on success.
    */
   @Retries.RetryTranslated
-  private void readSingleRangeWithRetries(
-      FileRange range, ByteBufferPool bufferPool) {
-    context.getReadInvoker().retry("vector read", uri, true, () ->
-            readSingleRange(range, bufferPool, false));
+  private IOException readSingleRangeWithRetries(
+      FileRange range,
+      ByteBufferPool bufferPool) {
+    try {
+      context.getReadInvoker().retry("vector read", uri, true, () ->
+              readSingleRange(range, bufferPool));
+      return null;
+    } catch (IOException ex) {
+      // the retry mechanism has stopped retrying, so mark the request as a failure.
+      range.getData().completeExceptionally(ex);
+      return ex;
+    }
   }
 
   /**
    * Read data from S3 for this range and populate a buffer.
+   * If the full read was succesful, the range's future is declared
+   * complete.
+   * <p>
+   * If an exception is raised,
+   * <ol>
+   *   <li>The buffer is returned to the pool.</li>
+   *   <li>The HTTP connection will be aborted if deemed to have failed.</li>
+   *   <li>The relevant statistics will be updated.</li>
+   *   <li>The exception is rethrown.</li>
+   * </ol>
+   * This is to allow the operation to be invoked in a retry() operation.
    * @param range range of data to read.
    * @param bufferPool buffer allocator.
-   * @param shouldRetry
+   * @throws IOException failure to GET or read the data.
    */
-  @Retries.RetryTranslated("GET is retried; reads are not")
-  private void readSingleRange(FileRange range, ByteBufferPool bufferPool, boolean shouldRetry) {
+  @Retries.OnceTranslated
+  private void readSingleRange(FileRange range,
+      ByteBufferPool bufferPool) throws IOException {
+
     LOG.debug("Start reading {} from {} ", range, pathStr);
     if (range.getLength() == 0) {
       ByteBuffer buffer = bufferPool.getBuffer(false, range.getLength());
@@ -1248,7 +1289,7 @@ private void readSingleRange(FileRange range, ByteBufferPool bufferPool, boolean
       long position = range.getOffset();
       int length = range.getLength();
       // a GET request, which has risk of failing if the file is gone, changed etc.
-      objectRange = getS3Object("readSingleRange", position, length, shouldRetry);
+      objectRange = getS3Object("readSingleRange", position, length, false);
 
       // GET has succeeded, make sure request is good to continue
       checkIfVectoredIOStopped();
@@ -1259,15 +1300,19 @@ private void readSingleRange(FileRange range, ByteBufferPool bufferPool, boolean
       populateBuffer(range, buffer, objectRange);
       range.getData().complete(buffer);
       LOG.debug("Finished reading range {} from path {}", range, pathStr);
-    } catch (Throwable ex) {
+    } catch (IOException ex) {
       // any failure.
       // log, the error, return the buffer to the pool, and report a failure.
       LOG.debug("Exception while reading a range {} from path {}", range, pathStr, ex);
       if (buffer != null) {
         // return any buffer to the pool
         bufferPool.putBuffer(buffer);
       }
-      range.getData().completeExceptionally(ex);
+      if (shouldInputStreamBeAborted(ex)) {
+        // abort the stream if the exception indicates this is needed.
+        abortSdkStream(uri, objectRange, streamStatistics, "read failure");
+      }
+      throw ex;
     } finally {
       IOUtils.cleanupWithLogger(LOG, objectRange);
     }
@@ -1280,7 +1325,7 @@ private void readSingleRange(FileRange range, ByteBufferPool bufferPool, boolean
    * @param range vector range to populate.
    * @param buffer buffer to fill.
    * @param objectContent result retrieved from S3 store.
-   * @throws IOException any IOE.
+   * @throws IOException any IOE raised reading the input stream.
    * @throws EOFException if EOF if read() call returns -1
    * @throws InterruptedIOException if vectored IO operation is stopped.
    */
@@ -1334,14 +1379,12 @@ private void readByteArray(InputStream objectContent,
               length - readBytes);
       LOG.debug("read {} bytes from stream", readBytesCurr);
       if (readBytesCurr < 0) {
-        // TODO: abort the stream.
         final String message = String.format("HTTP stream closed before all bytes were read."
                 + " Expected %,d bytes but only read %,d bytes. Current position %,d"
                 + " (%s)",
             length, readBytes, position, range);
         LOG.warn(message);
-        throw new EOFException(
-            message);
+        throw new EOFException(message);
       }
       readBytes += readBytesCurr;
       position += readBytesCurr;
@@ -1355,7 +1398,7 @@ private void readByteArray(InputStream objectContent,
    * Read data from S3 with retries for the GET request, as part of a vector IO
    * operation.
    * <p>
-   * This also handles if file has been changed while the
+   * This also handles the file being changed while the
    * http call is getting executed. If the file has been
    * changed RemoteFileChangedException is thrown.
    * <p>
@@ -1405,20 +1448,19 @@ private ResponseInputStream<GetObjectResponse> getS3Object(String operationName,
     }
     changeTracker.processResponse(objectRange.response(), operationName,
             position);
-    checkIfVectoredIOStopped();
     return objectRange;
   }
 
   /**
-   * Check if vectored io operation has been stooped. This happens
-   * when the stream is closed or unbuffer is called.
+   * Check if vectored io operation has been stopped. This happens
+   * when the stream is closed or unbuffer() was called during the read.
    * @throws InterruptedIOException throw InterruptedIOException such
    *                                that all running vectored io is
    *                                terminated thus releasing resources.
    */
   private void checkIfVectoredIOStopped() throws InterruptedIOException {
     if (stopVectoredIOOperations.get()) {
-      throw new InterruptedIOException("Stream closed or unbuffer is called");
+      throw new InterruptedIOException("Stream closed or unbuffer() was called during the read");
     }
   }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/ErrorTranslation.java

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 import java.lang.reflect.Constructor;
+import java.net.SocketTimeoutException;
 
 import software.amazon.awssdk.awscore.exception.AwsServiceException;
 import software.amazon.awssdk.core.exception.SdkException;
@@ -270,6 +271,17 @@ public static HttpChannelEOFException maybeExtractChannelException(
     return null;
   }
 
+  /**
+   * Is the exception to be considered as an unrecoverable channel failure
+   * -and that the stream should be aborted if so.
+   * @param t caught exception.
+   * @return true if the stream must not be returned to the pool.
+   */
+  public static boolean shouldInputStreamBeAborted(Throwable t) {
+    return (t instanceof HttpChannelEOFException)
+        || (t instanceof SocketTimeoutException);
+  }
+
   /**
    * AWS error codes explicitly recognized and processes specially;
    * kept in their own class for isolation.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/SDKStreamDrainer.java

@@ -21,6 +21,8 @@
 import java.io.InputStream;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import javax.annotation.Nullable;
+
 import software.amazon.awssdk.http.Abortable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,6 +40,7 @@
  * Drains/aborts s3 or other AWS SDK streams.
  * It is callable so can be passed directly to a submitter
  * for async invocation.
+ * @param <TStream> type of stream to drain/abort.
  */
 public class SDKStreamDrainer<TStream extends InputStream & Abortable>
     implements CallableRaisingIOE<Boolean> {
@@ -300,4 +303,31 @@ public String toString() {
         ", thrown=" + thrown +
         '}';
   }
+
+  /**
+   * Abort a stream, always.
+   * @param <TStream> type of stream to drain/abort.
+   * @param uri URI for messages
+   * @param sdkStream stream to close. Can be null.
+   * @param streamStatistics stats to update
+   * @param reason reason for stream being closed; used in messages
+   * @return true if the abort was successful.
+   */
+  public static <TStream extends InputStream & Abortable> boolean abortSdkStream(
+      final String uri,
+      @Nullable final TStream sdkStream,
+      final S3AInputStreamStatistics streamStatistics,
+      final String reason) {
+    if (sdkStream == null) {
+      return false;
+    }
+    return new SDKStreamDrainer<>(
+        uri,
+        sdkStream,
+        true,
+        0,
+        streamStatistics,
+        reason)
+        .apply();
+  }
 }