AVRO-3594: FsInput to use openFile() API

lang/java/mapred/src/main/java/org/apache/avro/mapred/FsInput.java

@@ -22,12 +22,17 @@
 import java.io.IOException;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FSDataInputStream;
 
 import org.apache.avro.file.SeekableInput;
 
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
+import static org.apache.hadoop.util.functional.FutureIO.awaitFuture;
+
 /** Adapt an {@link FSDataInputStream} to {@link SeekableInput}. */
 public class FsInput implements Closeable, SeekableInput {
   private final FSDataInputStream stream;
@@ -40,8 +45,19 @@ public FsInput(Path path, Configuration conf) throws IOException {
 
   /** Construct given a path and a {@code FileSystem}. */
   public FsInput(Path path, FileSystem fileSystem) throws IOException {
-    this.len = fileSystem.getFileStatus(path).getLen();
-    this.stream = fileSystem.open(path);
+    final FileStatus st = fileSystem.getFileStatus(path);
+    this.len = st.getLen();
+    // use the hadoop 3.3+ openFile API, passing in status
+    // and read policy. object stores can use these to
+    // optimize read performance and save on a HEAD request when opening
+    // a file.
+    final FutureDataInputStreamBuilder builder = fileSystem.openFile(path).opt(FS_OPTION_OPENFILE_READ_POLICY,
+        "avro, sequential, adaptive");
+    if (path.equals(st.getPath())) {
+      // set the file status if this isn't any wrapped filesystem.
+      builder.withFileStatus(st);
+    }
+    this.stream = awaitFuture(builder.build());
   }
 
   @Override

lang/java/mapred/src/test/java/org/apache/avro/mapreduce/TestFsInput.java

@@ -21,8 +21,10 @@
 import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
@@ -40,6 +42,7 @@
 public class TestFsInput {
   private static File file;
   private static final String FILE_CONTENTS = "abcdefghijklmnopqrstuvwxyz";
+  public static final int LENGTH = FILE_CONTENTS.length();
   private Configuration conf;
   private FsInput fsInput;
 
@@ -71,7 +74,7 @@ void configurationConstructor() throws Exception {
     try (FsInput in = new FsInput(new Path(file.getPath()), conf)) {
       int expectedByteCount = 1;
       byte[] readBytes = new byte[expectedByteCount];
-      int actualByteCount = fsInput.read(readBytes, 0, expectedByteCount);
+      int actualByteCount = in.read(readBytes, 0, expectedByteCount);
       assertThat(actualByteCount, is(equalTo(expectedByteCount)));
     }
   }
@@ -83,7 +86,7 @@ void fileSystemConstructor() throws Exception {
     try (FsInput in = new FsInput(path, fs)) {
       int expectedByteCount = 1;
       byte[] readBytes = new byte[expectedByteCount];
-      int actualByteCount = fsInput.read(readBytes, 0, expectedByteCount);
+      int actualByteCount = in.read(readBytes, 0, expectedByteCount);
       assertThat(actualByteCount, is(equalTo(expectedByteCount)));
     }
   }
@@ -123,4 +126,57 @@ void tell() throws Exception {
     assertThat(actualTellPos, is(equalTo(expectedTellPos)));
   }
 
+  /**
+   * How does a negative seek manifest itself? It's expected to fail on the seek()
+   * call and not move the file position. Most streams raise EOFExceotion; some
+   * raise IllegalArgumentException instead.
+   */
+  @Test
+  void seekNegative() throws Exception {
+    fsInput.seek(1);
+    assertThrows(Exception.class, () -> fsInput.seek(-1));
+    assertThat("file position after a negative seek", fsInput.tell(), is(equalTo(1L)));
+  }
+
+  /**
+   * Seek past the EOF then read.
+   */
+  @Test
+  void seekPastEOF() {
+    assertThrows(Exception.class, () -> fsInput.seek(LENGTH + 2));
+  }
+
+  /**
+   * Read to the exact EOF then do a read(), which is required to return -1 to
+   * indicate the EOF has been reached.
+   */
+  @Test
+  void readAtEOF() throws Exception {
+    fsInput.seek(LENGTH);
+    final int l = 8;
+    byte[] readBytes = new byte[l];
+    assertThat("bytes read from beyond EOF", fsInput.read(readBytes, 0, l), is(equalTo(-1)));
+  }
+
+  /**
+   * Read across the end of file. All data available up to the EOF is returned.
+   */
+  @Test
+  void readAcrossEOF() throws Exception {
+    fsInput.seek(LENGTH - 2);
+    final int l = 8;
+    byte[] readBytes = new byte[l];
+    assertThat("bytes read from beyond EOF", fsInput.read(readBytes, 0, l), is(equalTo(2)));
+    assertThat("bytes read from beyond EOF", fsInput.tell(), is(equalTo((long) LENGTH)));
+  }
+
+  /**
+   * Delete the file before trying to open it.
+   */
+  @Test
+  void openMissingFile() {
+    file.delete();
+    assertThrows(FileNotFoundException.class, () -> new FsInput(new Path(file.getPath()), conf).close());
+  }
+
 }