This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.1 by this push:
new 0c814bace6 Added VolumeManager.open with FileStatus parameter (#6484)
0c814bace6 is described below
commit 0c814bace66bb3f7471d534781146dbd1b786f2a
Author: Dave Marion <[email protected]>
AuthorDate: Mon Jul 20 16:53:10 2026 -0400
Added VolumeManager.open with FileStatus parameter (#6484)
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.
---
.../apache/accumulo/core/file/FileOperations.java | 34 +++++++++++++++
.../file/blockfile/impl/CachableBlockFile.java | 50 ++++++----------------
.../core/file/rfile/bcfile/PrintBCInfo.java | 9 ++--
.../apache/accumulo/server/fs/VolumeManager.java | 3 ++
.../accumulo/server/fs/VolumeManagerImpl.java | 6 +++
.../apache/accumulo/tserver/logger/LogReader.java | 8 ++--
6 files changed, 67 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 dd5b518d01..e3d4d492cd 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.file.rfile.RFile;
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 abstract class FileOperations {
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)
//
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 2a29023581..ba1c305bc5 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.io.UncheckedIOException;
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.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.FutureDataInputStreamBuilder;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Seekable;
import org.slf4j.Logger;
@@ -102,43 +100,21 @@ public class CachableBlockFile {
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();
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 fd8356a026..cf26aadac8 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 java.util.Set;
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();
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 6f7156d139..0921c00853 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 @@ public interface VolumeManager extends AutoCloseable {
// 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 8477db3be7..211e852915 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 java.util.stream.Stream;
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 class VolumeManagerImpl implements VolumeManager {
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 0d9ea4fd14..704a57804c 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;
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 class LogReader implements KeywordExecutable {
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 class LogReader implements KeywordExecutable {
}
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 {
