HADOOP-19043. S3A: Regression: ITestS3AOpenCost fails on prefetch tes…

…t runs

* Adds EOF logic deep into the prefetching code
* Tests still failing.

Change-Id: I9b23b01d010d8a1a680ce849d26a0aebab2389e2

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

@@ -29,14 +29,14 @@
 
 import org.apache.hadoop.fs.impl.prefetch.Validate;
 import org.apache.hadoop.fs.s3a.Invoker;
+import org.apache.hadoop.fs.s3a.Retries;
 import org.apache.hadoop.fs.s3a.S3AInputStream;
 import org.apache.hadoop.fs.s3a.S3AReadOpContext;
 import org.apache.hadoop.fs.s3a.S3AUtils;
 import org.apache.hadoop.fs.s3a.S3ObjectAttributes;
 import org.apache.hadoop.fs.s3a.impl.ChangeTracker;
 import org.apache.hadoop.fs.s3a.impl.SDKStreamDrainer;
 import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
-import org.apache.hadoop.fs.statistics.DurationTracker;
 
 import static org.apache.hadoop.fs.s3a.Invoker.onceTrackingDuration;
 
@@ -161,8 +161,6 @@ public long size() {
 
   /**
    * Opens a section of the file for reading.
-   * The s3 object response is cached in {@link #s3Objects} so that
-   * GC does not close the input stream.
    *
    * @param offset Start offset (0 based) of the section to read.
    * @param size Size of the section to read.
@@ -173,6 +171,7 @@ public long size() {
    * @throws IllegalArgumentException if offset is greater than or equal to file size.
    * @throws IllegalArgumentException if size is greater than the remaining bytes.
    */
+  @Retries.OnceTranslated
   public ResponseInputStream<GetObjectResponse> openForRead(long offset, int size)
       throws IOException {
     Validate.checkNotNegative(offset, "offset");
@@ -188,8 +187,7 @@ public ResponseInputStream<GetObjectResponse> openForRead(long offset, int size)
 
     String operation = String.format(
         "%s %s at %d size %d", S3AInputStream.OPERATION_OPEN, uri, offset, size);
-    DurationTracker tracker = streamStatistics.initiateGetRequest();
-    ResponseInputStream<GetObjectResponse> object = null;
+    ResponseInputStream<GetObjectResponse> object;
 
     // initiate the GET. This completes once the request returns the response headers;
     // the data is read later.

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

@@ -30,6 +30,8 @@
 import org.apache.hadoop.fs.impl.prefetch.Validate;
 import org.apache.hadoop.fs.s3a.HttpChannelEOFException;
 import org.apache.hadoop.fs.s3a.Invoker;
+import org.apache.hadoop.fs.s3a.RangeNotSatisfiableEOFException;
+import org.apache.hadoop.fs.s3a.Retries;
 import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
 
 import software.amazon.awssdk.core.ResponseInputStream;
@@ -78,13 +80,14 @@ public S3ARemoteObjectReader(S3ARemoteObject remoteObject) {
    * @param offset the absolute offset into the underlying file where reading starts.
    * @param size the number of bytes to be read.
    *
-   * @return number of bytes actually read.
+   * @return number of bytes actually read. -1 if the file is closed.
    * @throws IOException if there is an error reading from the file.
    *
    * @throws IllegalArgumentException if buffer is null.
    * @throws IllegalArgumentException if offset is outside of the range [0, file size].
    * @throws IllegalArgumentException if size is zero or negative.
    */
+  @Retries.RetryTranslated
   public int read(ByteBuffer buffer, long offset, int size) throws IOException {
     Validate.checkNotNull(buffer, "buffer");
     Validate.checkWithinRange(offset, "offset", 0, this.remoteObject.size());
@@ -103,31 +106,47 @@ public void close() {
     this.closed = true;
   }
 
+  /**
+   * Reads one block from S3.
+   * <p>
+   * There are no retries on base EOFExceptions.
+   * {@link HttpChannelEOFException} will be retried.
+   * {@link RangeNotSatisfiableEOFException} will be downgraded to
+   * partial read, so data may be returned.
+   * @param buffer destination.
+   * @param offset object offset
+   * @param size size to retrieve.
+   * @return bytes read.
+   * @throws EOFException if this was raised.
+   * @throws IOException IO failure.
+   */
+  @Retries.RetryTranslated
   private int readOneBlockWithRetries(ByteBuffer buffer, long offset, int size)
       throws IOException {
 
     this.streamStatistics.readOperationStarted(offset, size);
     Invoker invoker = this.remoteObject.getReadInvoker();
 
     final String path = this.remoteObject.getPath();
-    int invokerResponse =
+    EOFException invokerResponse =
         invoker.retry(String.format("read %s [%d-%d]", path, offset, size),
             path, true,
             trackDurationOfOperation(streamStatistics,
                 STREAM_READ_REMOTE_BLOCK_READ, () -> {
                   try {
                     this.readOneBlock(buffer, offset, size);
                   } catch (HttpChannelEOFException e) {
+                    // EOF subclasses with are rethrown as errors.
                     this.remoteObject.getStatistics().readException();
                     throw e;
                   } catch (EOFException e) {
                     // the base implementation swallows EOFs.
-                    return -1;
+                    return e;
                   } catch (IOException e) {
                     this.remoteObject.getStatistics().readException();
                     throw e;
                   }
-                  return 0;
+                  return null;
                 }));
 
     int numBytesRead = buffer.position();
@@ -139,13 +158,28 @@ private int readOneBlockWithRetries(ByteBuffer buffer, long offset, int size)
     this.remoteObject.getStatistics()
         .readOperationCompleted(size, numBytesRead);
 
-    if (invokerResponse < 0) {
-      return invokerResponse;
-    } else {
-      return numBytesRead;
+    if (invokerResponse != null) {
+      if (invokerResponse instanceof RangeNotSatisfiableEOFException) {
+        // the range wasn't satisfiable, but some may have been read.
+        return numBytesRead;
+      } else {
+        throw invokerResponse;
+      }
     }
+
+    // how much was read?
+    return numBytesRead;
   }
 
+  /**
+   * GET one block from S3.
+   * @param buffer buffer to fill up.
+   * @param offset offset within the object.
+   * @param size size to retrieve.
+   * @throws IOException IO failure.
+   * @throws HttpChannelEOFException if the channel is closed during the read.
+   */
+  @Retries.OnceTranslated
   private void readOneBlock(ByteBuffer buffer, long offset, int size)
       throws IOException {
     int readSize = Math.min(size, buffer.remaining());

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/AbstractS3ACostTest.java

@@ -370,6 +370,14 @@ protected OperationCostValidator.ExpectedProbe always(
     return expect(true, cost);
   }
 
+  /**
+   * Always run a metrics operation.
+   * @return a probe.
+   */
+  protected OperationCostValidator.ExpectedProbe always() {
+    return OperationCostValidator.always();
+  }
+
   /**
    * A metric diff which must hold when the fs is keeping markers.
    * @param cost expected cost

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestS3AOpenCost.java

@@ -41,8 +41,8 @@
 import org.apache.hadoop.fs.s3a.S3AInputStream;
 import org.apache.hadoop.fs.s3a.S3ATestUtils;
 import org.apache.hadoop.fs.s3a.Statistic;
-import org.apache.hadoop.fs.s3a.prefetch.S3APrefetchingInputStream;
 import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.io.IOUtils;
 
 import static org.apache.hadoop.fs.FSExceptionMessages.EOF_IN_READ_FULLY;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
@@ -153,26 +153,34 @@ public void testOpenFileWithStatusOfOtherFS() throws Throwable {
         new Path("gopher:///localhost/" + testFile.getName()));
 
     // no IO in open
-    FSDataInputStream in = verifyMetrics(() ->
-            fs.openFile(testFile)
-                .withFileStatus(st2)
-                .build()
-                .get(),
-        always(NO_HEAD_OR_LIST),
-        with(STREAM_READ_OPENED, 0));
-
-    // the stream gets opened during read
-    long readLen = verifyMetrics(() ->
-            readStream(in),
-        always(NO_HEAD_OR_LIST),
-        with(STREAM_READ_OPENED, 1));
-    assertEquals("bytes read from file", fileLength, readLen);
+    FSDataInputStream in = null;
+    try {
+      in = verifyMetrics(() ->
+              fs.openFile(testFile)
+                  .withFileStatus(st2)
+                  .build()
+                  .get(),
+          always(NO_HEAD_OR_LIST),
+          with(STREAM_READ_OPENED, 0));
+
+      // the stream gets opened during read
+      long readLen = verifyMetrics(() ->
+              readStream(in),
+          always(NO_HEAD_OR_LIST),
+          with(STREAM_READ_OPENED, 1));
+      assertEquals("bytes read from file", fileLength, readLen);
+    } finally {
+      IOUtils.closeStream(in);
+    }
   }
 
   @Test
   public void testStreamIsNotChecksummed() throws Throwable {
     describe("Verify that an opened stream is not checksummed");
-    S3AFileSystem fs = getFileSystem();
+
+    // if prefetching is enabled, skip this test
+    assumeNoPrefetching();S3AFileSystem fs = getFileSystem();
+
     // open the file
     try (FSDataInputStream in = verifyMetrics(() ->
             fs.openFile(testFile)
@@ -184,12 +192,6 @@ public void testStreamIsNotChecksummed() throws Throwable {
         always(NO_HEAD_OR_LIST),
         with(STREAM_READ_OPENED, 0))) {
 
-      // if prefetching is enabled, skip this test
-      final InputStream wrapped = in.getWrappedStream();
-      if (!(wrapped instanceof S3AInputStream)) {
-        skip("Not an S3AInputStream: " + wrapped);
-      }
-
       // open the stream.
       in.read();
       // now examine the innermost stream and make sure it doesn't have a checksum
@@ -259,6 +261,7 @@ public void testOpenFileLongerLengthReadFully() throws Throwable {
         return in.toString();
       }
     },
+        always(),
         // two GET calls were made, one for readFully,
         // the second on the read() past the EOF
         // the operation has got as far as S3
@@ -343,6 +346,7 @@ public void testReadPastEOF() throws Throwable {
         return in.toString();
       }
     },
+        always(),
         probe(!prefetching, Statistic.ACTION_HTTP_GET_REQUEST, extra));
   }
 
@@ -374,6 +378,7 @@ public void testPositionedReadableReadFullyPastEOF() throws Throwable {
             + ioStatisticsToPrettyString(in.getIOStatistics());
       }
     },
+        always(),
         probe(!prefetching, Statistic.ACTION_HTTP_GET_REQUEST, 1)); // no attempt to re-open
   }
 
@@ -408,6 +413,7 @@ public void testPositionedReadableReadPastEOF() throws Throwable {
         return "PositionedReadable.read()) past EOF with " + in;
       }
     },
+        always(),
         probe(!prefetching, Statistic.ACTION_HTTP_GET_REQUEST, 1)); // no attempt to re-open
   }
 
@@ -441,6 +447,7 @@ public void testVectorReadPastEOF() throws Throwable {
         return "vector read past EOF with " + in;
       }
     },
+        always(),
         probe(!prefetching, Statistic.ACTION_HTTP_GET_REQUEST, 1));
   }
 
@@ -454,6 +461,15 @@ private boolean prefetching() throws IOException {
         PREFETCH_ENABLED_KEY);
   }
 
+  /**
+   * Skip the test if prefetching is enabled.
+   */
+  private void assumeNoPrefetching(){
+    if (prefetching) {
+      skip("Prefetching is enabled");
+    }
+  }
+
   /**
    * Assert that the inner S3 Stream is closed.
    * @param in input stream

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/prefetch/ITestInMemoryInputStream.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.fs.s3a.prefetch;
 
+import org.assertj.core.api.Assertions;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,7 +56,7 @@
 
 /**
  * Test the prefetching input stream, validates that the
- * S3AInMemoryInputStream is working as expected.
+ * {@link S3AInMemoryInputStream} is working as expected.
  */
 public class ITestInMemoryInputStream extends AbstractS3ACostTest {
 
@@ -215,8 +216,9 @@ public void testStatusProbesAfterClosingStream() throws Throwable {
       assertEquals("Stream stats retrieved through stream before and after closing should match",
           inputStreamStatistics, newInputStreamStatistics);
 
-      assertFalse("seekToNewSource() not supported with prefetch", in.seekToNewSource(10));
-
+      Assertions.assertThat(in.seekToNewSource(10))
+          .describedAs("seekToNewSource() not supported with prefetch: %s", in)
+          .isFalse();
     }
 
   }