Added VolumeManager.open with FileStatus parameter

core/src/main/java/org/apache/accumulo/core/file/FileOperations.java

@@ -18,11 +18,15 @@
  */
 package org.apache.accumulo.core.file;
 
+import static java.util.Objects.requireNonNull;
 import static org.apache.accumulo.core.file.blockfile.impl.CacheProvider.NULL_PROVIDER;
 
 import java.io.IOException;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
@@ -35,9 +39,12 @@
 import org.apache.accumulo.core.spi.crypto.CryptoService;
 import org.apache.accumulo.core.util.ratelimit.RateLimiter;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.FileOutputCommitter;
 
 import com.google.common.cache.Cache;
@@ -76,6 +83,33 @@ public static FileOperations getInstance() {
     return new DispatchingFileFactory();
   }
 
+  public static FSDataInputStream openFile(FileSystem fs, Path path, FileStatus status)
+      throws IOException {
+    final FutureDataInputStreamBuilder builder = fs.openFile(path);
+    if (status != null) {
+      builder.withFileStatus(status);
+    }
+    final CompletableFuture<FSDataInputStream> future = builder.build();
+    while (!future.isDone()) {
+      try {
+        Thread.sleep(10);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw new IOException("Interrupted while opening file: " + path, e);
+      }
+    }
+    try {
+      return future.get();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new IOException("Interrupted while opening file: " + path, e);
+    } catch (CancellationException e) {
+      throw new IOException("Cancelled while opening file: " + path, e);
+    } catch (ExecutionException e) {
+      throw new IOException("Error trying to open file: " + path, e);
+    }
+  }
+
   //
   // Abstract methods (to be implemented by subclasses)
   //

core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java

@@ -26,12 +26,11 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
+import org.apache.accumulo.core.file.FileOperations;
 import org.apache.accumulo.core.file.rfile.BlockIndex;
 import org.apache.accumulo.core.file.rfile.bcfile.BCFile;
 import org.apache.accumulo.core.file.rfile.bcfile.BCFile.Reader.BlockReader;
@@ -49,7 +48,6 @@
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.Seekable;
 import org.slf4j.Logger;
@@ -102,43 +100,21 @@ public CachableBuilder fsPath(FileSystem fs, Path dataFile, boolean dropCacheBeh
         FileStatus status) {
       this.cacheId = pathToCacheId(dataFile);
       this.inputSupplier = () -> {
-        FutureDataInputStreamBuilder builder = fs.openFile(dataFile);
-        if (status != null) {
-          builder.withFileStatus(status);
-        }
-        CompletableFuture<FSDataInputStream> future = builder.build();
-        while (!future.isDone()) {
+        FSDataInputStream is = FileOperations.openFile(fs, dataFile, status);
+        if (dropCacheBehind) {
+          // Tell the DataNode that the write ahead log does not need to be cached in the OS page
+          // cache
           try {
-            Thread.sleep(10);
-          } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            throw new IOException("Interrupted while opening file: " + dataFile, e);
-          }
-        }
-        try {
-          FSDataInputStream is = future.get();
-          if (dropCacheBehind) {
-            // Tell the DataNode that the write ahead log does not need to be cached in the OS page
-            // cache
-            try {
-              is.setDropBehind(Boolean.TRUE);
-              log.trace("Called setDropBehind(TRUE) for stream reading file {}", dataFile);
-            } catch (UnsupportedOperationException e) {
-              log.debug("setDropBehind not enabled for wal file: {}", dataFile);
-            } catch (IOException e) {
-              log.debug("IOException setting drop behind for file: {}, msg: {}", dataFile,
-                  e.getMessage());
-            }
+            is.setDropBehind(Boolean.TRUE);
+            log.trace("Called setDropBehind(TRUE) for stream reading file {}", dataFile);
+          } catch (UnsupportedOperationException e) {
+            log.debug("setDropBehind not enabled for wal file: {}", dataFile);
+          } catch (IOException e) {
+            log.debug("IOException setting drop behind for file: {}, msg: {}", dataFile,
+                e.getMessage());
           }
-          return is;
-        } catch (InterruptedException e) {
-          Thread.currentThread().interrupt();
-          throw new IOException("Interrupted while opening file: " + dataFile, e);
-        } catch (CancellationException e) {
-          throw new IOException("Cancelled while opening file: " + dataFile, e);
-        } catch (ExecutionException e) {
-          throw new IOException("Error trying to open file: " + dataFile, e);
         }
+        return is;
       };
       this.lengthSupplier =
           () -> status == null ? fs.getFileStatus(dataFile).getLen() : status.getLen();

core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintBCInfo.java

@@ -25,11 +25,13 @@
 
 import org.apache.accumulo.core.cli.ConfigOpts;
 import org.apache.accumulo.core.conf.SiteConfiguration;
+import org.apache.accumulo.core.file.FileOperations;
 import org.apache.accumulo.core.file.rfile.bcfile.BCFile.MetaIndexEntry;
 import org.apache.accumulo.core.spi.crypto.CryptoService;
 import org.apache.accumulo.core.spi.crypto.NoCryptoServiceFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 
@@ -47,9 +49,10 @@ public class PrintBCInfo {
   CryptoService cryptoService = NoCryptoServiceFactory.NONE;
 
   public void printMetaBlockInfo() throws IOException {
-    FSDataInputStream fsin = fs.open(path);
-    try (BCFile.Reader bcfr =
-        new BCFile.Reader(fsin, fs.getFileStatus(path).getLen(), conf, cryptoService)) {
+    FileStatus status = fs.getFileStatus(path);
+
+    try (FSDataInputStream fsin = FileOperations.openFile(fs, path, status);
+        BCFile.Reader bcfr = new BCFile.Reader(fsin, status.getLen(), conf, cryptoService)) {
 
       Set<Entry<String,MetaIndexEntry>> es = bcfr.metaIndex.index.entrySet();
 

server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java

@@ -157,6 +157,9 @@ RemoteIterator<LocatedFileStatus> listFiles(final Path path, final boolean recur
   // forward to the appropriate FileSystem object
   FSDataInputStream open(Path path) throws IOException;
 
+  // forward to the appropriate FileSystem object
+  FSDataInputStream open(Path path, FileStatus status) throws IOException;
+
   // forward to the appropriate FileSystem object, throws an exception if the paths are in different
   // volumes
   boolean rename(Path path, Path newPath) throws IOException;

server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java

@@ -44,6 +44,7 @@
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.DefaultConfiguration;
 import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.file.FileOperations;
 import org.apache.accumulo.core.spi.fs.VolumeChooser;
 import org.apache.accumulo.core.util.Pair;
 import org.apache.accumulo.core.util.threads.ThreadPools;
@@ -308,6 +309,11 @@ public FSDataInputStream open(Path path) throws IOException {
     return getFileSystemByPath(path).open(path);
   }
 
+  @Override
+  public FSDataInputStream open(Path path, FileStatus status) throws IOException {
+    return FileOperations.openFile(getFileSystemByPath(path), path, status);
+  }
+
   @Override
   public boolean rename(Path path, Path newPath) throws IOException {
     FileSystem source = getFileSystemByPath(path);

server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java

@@ -49,6 +49,7 @@
 import org.apache.accumulo.tserver.log.DfsLogger.LogHeaderIncompleteException;
 import org.apache.accumulo.tserver.log.RecoveryLogsIterator;
 import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
 import org.slf4j.Logger;
@@ -140,21 +141,22 @@ public void execute(String[] args) throws Exception {
         LogFileValue value = new LogFileValue();
 
         // ensure it's a regular non-sorted WAL file, and not a single sorted WAL in RFile format
-        if (fs.getFileStatus(path).isFile()) {
+        FileStatus status = fs.getFileStatus(path);
+        if (status.isFile()) {
           if (file.endsWith(".rf")) {
             log.error("Unable to read from a single RFile. A non-sorted WAL file was expected. "
                 + "To read sorted WALs, please pass in a directory containing the sorted recovery logs.");
             continue;
           }
 
           if (opts.printOnlyEncryptionInfo) {
-            try (final FSDataInputStream fsinput = fs.open(path)) {
+            try (final FSDataInputStream fsinput = fs.open(path, status)) {
               printCryptoParams(fsinput, path);
             }
             continue;
           }
 
-          try (final FSDataInputStream fsinput = fs.open(path);
+          try (final FSDataInputStream fsinput = fs.open(path, status);
               DataInputStream input = DfsLogger.getDecryptingStream(fsinput, walCryptoService)) {
             while (true) {
               try {