From c98f0e8dd4bbb0f6a04c972bd02d5a84bc9156de Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Mon, 20 Jul 2026 19:59:58 +0000 Subject: [PATCH 1/2] Added VolumeManager.open method that uses FileStatus to reduce NN ops PR #6460 modified the code that opens RFiles to use a method that uses a FileStatus to reduce the number of RPCs to the NameNode. There are a couple of places in the code where this case exists, but the files are not RFiles. This change adds a similar method to the VolumeManager and moves the file open logic to FileOperations. --- .../accumulo/core/file/FileOperations.java | 34 +++++++++++++ .../blockfile/impl/CachableBlockFile.java | 50 +++++-------------- .../core/file/rfile/bcfile/PrintBCInfo.java | 8 +-- .../accumulo/server/fs/VolumeManager.java | 3 ++ .../accumulo/server/fs/VolumeManagerImpl.java | 6 +++ .../accumulo/tserver/logger/LogReader.java | 8 +-- 6 files changed, 66 insertions(+), 43 deletions(-) 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..bf8ca1907de 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,9 @@ 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); + FSDataInputStream fsin = FileOperations.openFile(fs, path, status); + try (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 { From e0f982b2326b4266f8de9fb9e3c73975c13a1c05 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Mon, 20 Jul 2026 20:09:14 +0000 Subject: [PATCH 2/2] Moved input stream into try-with-resources --- .../apache/accumulo/core/file/rfile/bcfile/PrintBCInfo.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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 bf8ca1907de..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 @@ -50,8 +50,9 @@ public class PrintBCInfo { public void printMetaBlockInfo() throws IOException { FileStatus status = fs.getFileStatus(path); - FSDataInputStream fsin = FileOperations.openFile(fs, path, status); - try (BCFile.Reader bcfr = new BCFile.Reader(fsin, status.getLen(), conf, cryptoService)) { + + 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();