diff --git a/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java b/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java index dd5b518d01e..e3d4d492cd6 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java +++ b/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 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) // diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java index 2a290235811..ba1c305bc55 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java +++ b/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 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(); diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintBCInfo.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintBCInfo.java index fd8356a0261..cf26aadac82 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/PrintBCInfo.java +++ b/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> es = bcfr.metaIndex.index.entrySet(); diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java index 6f7156d1391..0921c008539 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java +++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java @@ -157,6 +157,9 @@ RemoteIterator 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; diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java index 8477db3be7a..211e852915e 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java +++ b/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); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java index 0d9ea4fd142..704a57804c6 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java +++ b/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,7 +141,8 @@ 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."); @@ -148,13 +150,13 @@ public void execute(String[] args) throws Exception { } 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 {