This is an automated email from the ASF dual-hosted git repository.
steveloughran pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/trunk by this push:
new 62daf8933ad HDFS-17593. Allow setting block locations when opening
streams (#8561)
62daf8933ad is described below
commit 62daf8933ad7c66f463eb4eb2ac2dafef4b5186a
Author: boroknagyz <[email protected]>
AuthorDate: Sat Jun 27 12:25:18 2026 +0200
HDFS-17593. Allow setting block locations when opening streams (#8561)
Adds support for opening HDFS files with pre-fetched block locations,
eliminating redundant NameNode RPCs when callers already have cached
block metadata.
To achieve this, use the openFile()/openPath() builder methods and pass
in a filestatus; if it is an HdfsFileStatus the locations are picked up.
Contributed by Zoltan Borok-Nagy , with assistance Claude Opus 4.6
---
.../java/org/apache/hadoop/hdfs/DFSClient.java | 20 ++
.../apache/hadoop/hdfs/DistributedFileSystem.java | 90 +++++
.../src/main/native/libhdfs/hdfs.c | 73 ++++
.../src/main/native/libhdfs/include/hdfs/hdfs.h | 22 +-
.../src/main/native/libhdfspp/tests/hdfs_shim.c | 15 +
.../libhdfspp/tests/libhdfs_wrapper_defines.h | 1 +
.../libhdfspp/tests/libhdfs_wrapper_undefs.h | 1 +
.../libhdfspp/tests/libhdfspp_wrapper_defines.h | 1 +
.../hadoop/hdfs/TestOpenFileWithLocatedBlocks.java | 394 +++++++++++++++++++++
9 files changed, 614 insertions(+), 3 deletions(-)
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
index cbe7516b0ed..9a2b0e5fd5c 100755
---
a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
+++
b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
@@ -1078,6 +1078,26 @@ public DFSInputStream open(String src, int buffersize,
boolean verifyChecksum)
}
}
+ /**
+ * Create an input stream using pre-fetched block locations, skipping the
+ * NameNode RPC to get block locations.
+ * @param src file name
+ * @param buffersize ignored
+ * @param verifyChecksum verify checksums before returning data to client
+ * @param locatedBlocks pre-fetched block locations for this file
+ * @return an input stream for reading the file
+ * @throws IOException on I/O error
+ */
+ public DFSInputStream open(String src, int buffersize, boolean
verifyChecksum,
+ LocatedBlocks locatedBlocks) throws IOException {
+ Preconditions.checkArgument(locatedBlocks != null,
+ "null locatedBlocks");
+ checkOpen();
+ try (TraceScope ignored = newPathTraceScope("newDFSInputStream", src)) {
+ return openInternal(locatedBlocks, src, verifyChecksum);
+ }
+ }
+
/**
* Create an input stream from the {@link HdfsPathHandle} if the
* constraints encoded from {@link
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
index dac205158d0..dd8b35de905 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
+++
b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
@@ -69,7 +69,9 @@
import org.apache.hadoop.fs.UnresolvedLinkException;
import org.apache.hadoop.fs.UnsupportedFileSystemException;
import org.apache.hadoop.fs.XAttrSetFlag;
+import org.apache.hadoop.fs.impl.AbstractFSBuilderImpl;
import org.apache.hadoop.fs.impl.FileSystemMultipartUploaderBuilder;
+import org.apache.hadoop.fs.impl.OpenFileParameters;
import org.apache.hadoop.fs.permission.AclEntry;
import org.apache.hadoop.fs.permission.AclStatus;
import org.apache.hadoop.fs.permission.FsAction;
@@ -115,6 +117,7 @@
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.DelegationTokenIssuer;
+import org.apache.hadoop.util.LambdaUtils;
import org.apache.hadoop.util.Lists;
import org.apache.hadoop.util.Progressable;
import org.slf4j.Logger;
@@ -123,6 +126,8 @@
import javax.annotation.Nonnull;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.CompletableFuture;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.ArrayList;
@@ -378,6 +383,91 @@ public FSDataInputStream open(PathHandle fd, int
bufferSize)
return dfs.createWrappedInputStream(dfsis);
}
+ @Override
+ protected CompletableFuture<FSDataInputStream> openFileWithOptions(
+ final Path path,
+ final OpenFileParameters parameters) throws IOException {
+ AbstractFSBuilderImpl.rejectUnknownMandatoryKeys(
+ parameters.getMandatoryKeys(),
+ Options.OpenFileOptions.FS_OPTION_OPENFILE_STANDARD_OPTIONS,
+ "for " + path);
+ statistics.incrementReadOps(1);
+ storageStatistics.incrementOpCounter(OpType.OPEN);
+ final Path absF = fixRelativePart(path);
+ return LambdaUtils.eval(new CompletableFuture<>(), () -> {
+ LocatedBlocks locatedBlocks =
+ getLocatedBlocksFromStatus(absF, parameters.getStatus());
+ final DFSInputStream dfsis;
+ if (locatedBlocks != null) {
+ dfsis = dfs.open(getPathName(absF), parameters.getBufferSize(),
+ verifyChecksum, locatedBlocks);
+ } else {
+ dfsis = dfs.open(getPathName(absF), parameters.getBufferSize(),
+ verifyChecksum);
+ }
+ try {
+ return dfs.createWrappedInputStream(dfsis);
+ } catch (IOException e) {
+ dfsis.close();
+ throw e;
+ }
+ });
+ }
+
+ private static LocatedBlocks getLocatedBlocksFromStatus(
+ Path path, FileStatus status) {
+ if (status instanceof HdfsLocatedFileStatus locatedFileStatus) {
+ Preconditions.checkArgument(
+ path.getName().equals(status.getPath().getName()),
+ "Filename mismatch between file being opened %s"
+ + " and supplied status %s",
+ path, status.getPath());
+ return locatedFileStatus.getLocatedBlocks();
+ }
+ return null;
+ }
+
+ /**
+ * Create a new input stream for the same file as an existing stream,
+ * reusing its cached block locations to avoid a NameNode RPC.
+ * The returned stream is independent (its own position, buffers, etc.)
+ * but shares the same block location metadata.
+ *
+ * @param existing an open input stream obtained from this filesystem
+ * @return a new independent input stream for the same file
+ * @throws IOException if the stream cannot be cloned
+ */
+ public FSDataInputStream cloneDataInputStream(FSDataInputStream existing)
+ throws IOException {
+ statistics.incrementReadOps(1);
+ storageStatistics.incrementOpCounter(OpType.OPEN);
+ InputStream wrapped = existing.getWrappedStream();
+ DFSInputStream dfsis;
+ if (wrapped instanceof DFSInputStream dfsIn) {
+ dfsis = dfsIn;
+ } else if (wrapped instanceof
+ org.apache.hadoop.crypto.CryptoInputStream cryptoIn) {
+ dfsis = (DFSInputStream) cryptoIn.getWrappedStream();
+ } else {
+ throw new IOException("Cannot clone: underlying stream is "
+ + wrapped.getClass().getName() + ", not a DFSInputStream");
+ }
+ if (dfsis.getDFSClient() != dfs) {
+ throw new IOException(
+ "Cannot clone: stream belongs to a different DFSClient instance");
+ }
+ LocatedBlocks locatedBlocks = dfsis.getLocatedBlocks();
+ String src = dfsis.getSrc();
+ DFSInputStream clone = dfs.open(src,
+ dfs.getConf().getIoBufferSize(), verifyChecksum, locatedBlocks);
+ try {
+ return dfs.createWrappedInputStream(clone);
+ } catch (IOException e) {
+ clone.close();
+ throw e;
+ }
+ }
+
@Override
public String getErasureCodingPolicyName(FileStatus fileStatus) {
if (!(fileStatus instanceof HdfsFileStatus)) {
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/hdfs.c
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/hdfs.c
index 20cfb7a2c5d..6dcfa1264c5 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/hdfs.c
+++
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/hdfs.c
@@ -1885,6 +1885,79 @@ int hdfsCloseFile(hdfsFS fs, hdfsFile file)
return 0;
}
+hdfsFile hdfsCloneFile(hdfsFS fs, hdfsFile file)
+{
+ // JAVA EQUIVALENT:
+ // ((DistributedFileSystem) fs).cloneDataInputStream(file)
+ int ret = 0;
+ jthrowable jthr;
+ jvalue jVal;
+ jobject jFS = (jobject)fs;
+ jobject jClonedStream = NULL;
+ hdfsFile clonedFile = NULL;
+
+ JNIEnv* env = getJNIEnv();
+ if (env == NULL) {
+ errno = EINTERNAL;
+ return NULL;
+ }
+
+ if (!file || file->type != HDFS_STREAM_INPUT) {
+ fprintf(stderr, "hdfsCloneFile: can only clone input streams\n");
+ errno = EINVAL;
+ return NULL;
+ }
+
+ if (!(*env)->IsInstanceOf(env, jFS,
+ getJclass(JC_DISTRIBUTED_FILE_SYSTEM))) {
+ fprintf(stderr,
+ "hdfsCloneFile: not a DistributedFileSystem\n");
+ errno = ENOTSUP;
+ return NULL;
+ }
+
+ jthr = invokeMethod(env, &jVal, INSTANCE, jFS,
+ JC_DISTRIBUTED_FILE_SYSTEM, "cloneDataInputStream",
+ JMETHOD1(JPARAM(HADOOP_FSDISTRM), JPARAM(HADOOP_FSDISTRM)),
+ file->file);
+ if (jthr) {
+ ret = printExceptionAndFree(env, jthr, PRINT_EXC_ALL,
+ "hdfsCloneFile: DistributedFileSystem#cloneDataInputStream");
+ goto done;
+ }
+ jClonedStream = jVal.l;
+
+ clonedFile = calloc(1, sizeof(struct hdfsFile_internal));
+ if (!clonedFile) {
+ fprintf(stderr, "hdfsCloneFile: OOM\n");
+ ret = ENOMEM;
+ goto done;
+ }
+ clonedFile->file = (*env)->NewGlobalRef(env, jClonedStream);
+ if (!clonedFile->file) {
+ ret = printPendingExceptionAndFree(env, PRINT_EXC_ALL,
+ "hdfsCloneFile: NewGlobalRef");
+ goto done;
+ }
+ clonedFile->type = HDFS_STREAM_INPUT;
+ clonedFile->flags = 0;
+ setFileFlagCapabilities(clonedFile, jClonedStream);
+
+done:
+ destroyLocalReference(env, jClonedStream);
+ if (ret) {
+ if (clonedFile) {
+ if (clonedFile->file) {
+ (*env)->DeleteGlobalRef(env, clonedFile->file);
+ }
+ free(clonedFile);
+ }
+ errno = ret;
+ return NULL;
+ }
+ return clonedFile;
+}
+
int hdfsExists(hdfsFS fs, const char *path)
{
JNIEnv *env = getJNIEnv();
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/include/hdfs/hdfs.h
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/include/hdfs/hdfs.h
index eba50ff6eb2..62be5a967fa 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/include/hdfs/hdfs.h
+++
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/include/hdfs/hdfs.h
@@ -685,11 +685,27 @@ extern "C" {
int hdfsCloseFile(hdfsFS fs, hdfsFile file);
- /**
- * hdfsExists - Checks if a given path exsits on the filesystem
+ /**
+ * hdfsCloneFile - Create a new independent file handle for the same file
+ * as an existing handle, reusing cached block locations to avoid a
+ * NameNode RPC. The cloned handle has its own read position and buffers.
+ * Only works for input (read) streams on a DistributedFileSystem.
+ *
+ * @param fs The configured filesystem handle.
+ * @param file An open input file handle to clone.
+ * @return Returns a new file handle on success, NULL on error.
+ * On error, errno will be set appropriately.
+ * The returned handle must be closed with hdfsCloseFile.
+ */
+ LIBHDFS_EXTERNAL
+ hdfsFile hdfsCloneFile(hdfsFS fs, hdfsFile file);
+
+
+ /**
+ * hdfsExists - Checks if a given path exists on the filesystem
* @param fs The configured filesystem handle.
* @param path The path to look for
- * @return Returns 0 on success, -1 on error.
+ * @return Returns 0 on success, -1 on error.
*/
LIBHDFS_EXTERNAL
int hdfsExists(hdfsFS fs, const char *path);
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_shim.c
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_shim.c
index ad8ad712c9d..61c3b81cd6a 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_shim.c
+++
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_shim.c
@@ -331,6 +331,21 @@ int hdfsCloseFile(hdfsFS fs, hdfsFile file) {
return ret;
}
+hdfsFile hdfsCloneFile(hdfsFS fs, hdfsFile file) {
+ libhdfs_hdfsFile clonedRep =
+ libhdfs_hdfsCloneFile(fs->libhdfsRep, file->libhdfsRep);
+ if (!clonedRep) {
+ return NULL;
+ }
+ hdfsFile cloned = calloc(1, sizeof(struct hdfsFile_internal));
+ if (!cloned) {
+ return NULL;
+ }
+ cloned->libhdfsRep = clonedRep;
+ cloned->libhdfsppRep = NULL;
+ return cloned;
+}
+
int hdfsExists(hdfsFS fs, const char *path) {
return libhdfspp_hdfsExists(fs->libhdfsppRep, path);
}
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfs_wrapper_defines.h
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfs_wrapper_defines.h
index 16574414255..c9f8dfe38ff 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfs_wrapper_defines.h
+++
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfs_wrapper_defines.h
@@ -59,6 +59,7 @@
#define hdfsTruncateFile libhdfs_hdfsTruncateFile
#define hdfsUnbufferFile libhdfs_hdfsUnbufferFile
#define hdfsCloseFile libhdfs_hdfsCloseFile
+#define hdfsCloneFile libhdfs_hdfsCloneFile
#define hdfsExists libhdfs_hdfsExists
#define hdfsSeek libhdfs_hdfsSeek
#define hdfsTell libhdfs_hdfsTell
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfs_wrapper_undefs.h
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfs_wrapper_undefs.h
index d84b8ba2875..5f6aa3aa1c9 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfs_wrapper_undefs.h
+++
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfs_wrapper_undefs.h
@@ -59,6 +59,7 @@
#undef hdfsTruncateFile
#undef hdfsUnbufferFile
#undef hdfsCloseFile
+#undef hdfsCloneFile
#undef hdfsExists
#undef hdfsSeek
#undef hdfsTell
diff --git
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfspp_wrapper_defines.h
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfspp_wrapper_defines.h
index 0a6d987409f..da775c054a7 100644
---
a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfspp_wrapper_defines.h
+++
b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/libhdfspp_wrapper_defines.h
@@ -59,6 +59,7 @@
#define hdfsTruncateFile libhdfspp_hdfsTruncateFile
#define hdfsUnbufferFile libhdfspp_hdfsUnbufferFile
#define hdfsCloseFile libhdfspp_hdfsCloseFile
+#define hdfsCloneFile libhdfspp_hdfsCloneFile
#define hdfsExists libhdfspp_hdfsExists
#define hdfsSeek libhdfspp_hdfsSeek
#define hdfsTell libhdfspp_hdfsTell
diff --git
a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestOpenFileWithLocatedBlocks.java
b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestOpenFileWithLocatedBlocks.java
new file mode 100644
index 00000000000..66fe7db497f
--- /dev/null
+++
b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestOpenFileWithLocatedBlocks.java
@@ -0,0 +1,394 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs;
+
+import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.crypto.CryptoInputStream;
+import org.apache.hadoop.crypto.key.JavaKeyStoreProvider;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystemTestHelper;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.hdfs.client.CreateEncryptionZoneFlag;
+import org.apache.hadoop.hdfs.client.HdfsAdmin;
+import org.apache.hadoop.hdfs.protocol.HdfsLocatedFileStatus;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that opening a file with pre-fetched block locations via the
+ * {@code openFile().withFileStatus()} builder API skips the NameNode RPC
+ * for block locations.
+ */
+public class TestOpenFileWithLocatedBlocks {
+ private static final String NN_METRICS = "NameNodeActivity";
+ private static final int BLOCK_SIZE = 1024;
+ private static final short REPLICATION = 1;
+ private static final String TEST_KEY = "test_key";
+ private static final EnumSet<CreateEncryptionZoneFlag> NO_TRASH =
+ EnumSet.of(CreateEncryptionZoneFlag.NO_TRASH);
+
+ private static MiniDFSCluster cluster;
+ private static DistributedFileSystem fs;
+ private static Configuration conf;
+ private static Path ezPath;
+
+ @BeforeAll
+ public static void setupCluster() throws Exception {
+ conf = new HdfsConfiguration();
+ conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, BLOCK_SIZE);
+ FileSystemTestHelper fsHelper = new FileSystemTestHelper();
+ File testRootDir = new File(fsHelper.getTestRootDir()).getAbsoluteFile();
+ conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_KEY_PROVIDER_PATH,
+ JavaKeyStoreProvider.SCHEME_NAME + "://file" +
+ new Path(testRootDir.toString(), "test.jks").toUri());
+ cluster = new MiniDFSCluster.Builder(conf)
+ .numDataNodes(1)
+ .build();
+ cluster.waitActive();
+ fs = cluster.getFileSystem();
+ createEncryptionZone();
+ }
+
+ private static void createEncryptionZone() throws Exception {
+ DFSTestUtil.createKey(TEST_KEY, cluster, conf);
+ fs.getClient().setKeyProvider(
+ cluster.getNameNode().getNamesystem().getProvider());
+ ezPath = new Path("/ez");
+ fs.mkdirs(ezPath);
+ new HdfsAdmin(cluster.getURI(), conf)
+ .createEncryptionZone(ezPath, TEST_KEY, NO_TRASH);
+ }
+
+ @AfterAll
+ public static void teardownCluster() {
+ if (cluster != null) {
+ cluster.shutdown();
+ cluster = null;
+ }
+ }
+
+ private Path createFile(String name, byte[] data) throws IOException {
+ Path path = new Path("/" + name);
+ ContractTestUtils.createFile(fs, path, true, data);
+ return path;
+ }
+
+ @Test
+ public void testOpenFileWithLocatedFileStatus() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 2, 'a', 26);
+ Path path = createFile("testfile", data);
+
+ RemoteIterator<LocatedFileStatus> iter = fs.listLocatedStatus(path);
+ HdfsLocatedFileStatus locatedStatus =
+ (HdfsLocatedFileStatus) iter.next();
+
+ long countBefore = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+
+ try (FSDataInputStream in = fs.openFile(path)
+ .withFileStatus(locatedStatus)
+ .build()
+ .get()) {
+ ContractTestUtils.verifyRead(in, data, 0, data.length);
+ }
+
+ long countAfter = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+ assertEquals(countBefore, countAfter,
+ "Opening with pre-fetched locations should not trigger " +
+ "GetBlockLocations RPC");
+ }
+
+ @Test
+ public void testOpenFileWithoutStatusFallsBack() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE, 'b', 26);
+ Path path = createFile("testfile_no_status", data);
+
+ long countBefore = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+
+ try (FSDataInputStream in = fs.openFile(path)
+ .build()
+ .get()) {
+ ContractTestUtils.verifyRead(in, data, 0, data.length);
+ }
+
+ long countAfter = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+ assertEquals(countBefore + 1, countAfter,
+ "Opening without status should trigger GetBlockLocations RPC");
+ }
+
+ @Test
+ public void testOpenFileWithPlainFileStatusFallsBack() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE, 'c', 26);
+ Path path = createFile("testfile_plain_status", data);
+
+ FileStatus plainStatus = fs.getFileStatus(path);
+
+ long countBefore = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+
+ try (FSDataInputStream in = fs.openFile(path)
+ .withFileStatus(plainStatus)
+ .build()
+ .get()) {
+ ContractTestUtils.verifyRead(in, data, 0, data.length);
+ }
+
+ long countAfter = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+ assertEquals(countBefore + 1, countAfter,
+ "Opening with plain FileStatus should trigger " +
+ "GetBlockLocations RPC");
+ }
+
+ @Test
+ public void testCloneDataInputStream() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 3, 'd', 26);
+ Path path = createFile("testfile_clone", data);
+
+ try (FSDataInputStream original = fs.open(path)) {
+ long countBefore = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+
+ try (FSDataInputStream cloned = fs.cloneDataInputStream(original)) {
+ long countAfter = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+ assertEquals(countBefore, countAfter,
+ "Cloning should not trigger GetBlockLocations RPC");
+
+ ContractTestUtils.verifyRead(cloned, data, 0, data.length);
+ }
+
+ ContractTestUtils.verifyRead(original, data, 0, data.length);
+ }
+ }
+
+ @Test
+ public void testCloneDataInputStreamIndependentPosition() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 2, 'e', 26);
+ Path path = createFile("testfile_clone_pos", data);
+
+ try (FSDataInputStream original = fs.open(path)) {
+ byte[] buf1 = new byte[BLOCK_SIZE];
+ original.readFully(buf1);
+
+ try (FSDataInputStream cloned = fs.cloneDataInputStream(original)) {
+ assertEquals(0, cloned.getPos());
+ assertEquals(BLOCK_SIZE, original.getPos());
+
+ ContractTestUtils.verifyRead(cloned, data, 0, data.length);
+ }
+ }
+ }
+
+ @Test
+ public void testMultipleClonesFromSameHandle() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 2, 'f', 26);
+ Path path = createFile("testfile_multi_clone", data);
+
+ try (FSDataInputStream original = fs.open(path)) {
+ long countBefore = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+
+ int numClones = 5;
+ FSDataInputStream[] clones = new FSDataInputStream[numClones];
+ for (int i = 0; i < numClones; i++) {
+ clones[i] = fs.cloneDataInputStream(original);
+ }
+
+ long countAfter = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+ assertEquals(countBefore, countAfter,
+ "Multiple clones should not trigger GetBlockLocations RPC");
+
+ for (int i = 0; i < numClones; i++) {
+ ContractTestUtils.verifyRead(clones[i], data, 0, data.length);
+ clones[i].close();
+ }
+ }
+ }
+
+ @Test
+ public void testCloneOfClone() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 2, 'g', 26);
+ Path path = createFile("testfile_clone_of_clone", data);
+
+ try (FSDataInputStream original = fs.open(path)) {
+ long countBefore = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+
+ try (FSDataInputStream clone1 = fs.cloneDataInputStream(original);
+ FSDataInputStream clone2 = fs.cloneDataInputStream(clone1)) {
+ long countAfter = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+ assertEquals(countBefore, countAfter,
+ "Clone of clone should not trigger GetBlockLocations RPC");
+
+ ContractTestUtils.verifyRead(clone2, data, 0, data.length);
+ }
+ }
+ }
+
+ @Test
+ public void testConcurrentPreadFromClones() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 4, 'h', 26);
+ Path path = createFile("testfile_concurrent", data);
+
+ try (FSDataInputStream original = fs.open(path)) {
+ int numReaders = 4;
+ FSDataInputStream[] clones = new FSDataInputStream[numReaders];
+ for (int i = 0; i < numReaders; i++) {
+ clones[i] = fs.cloneDataInputStream(original);
+ }
+
+ ExecutorService executor = Executors.newFixedThreadPool(numReaders);
+ List<Future<byte[]>> futures = new ArrayList<>();
+ for (int i = 0; i < numReaders; i++) {
+ final int blockIdx = i;
+ final FSDataInputStream stream = clones[i];
+ futures.add(executor.submit(() -> {
+ byte[] buf = new byte[BLOCK_SIZE];
+ stream.readFully(blockIdx * BLOCK_SIZE, buf);
+ return buf;
+ }));
+ }
+
+ for (int i = 0; i < numReaders; i++) {
+ byte[] expected = new byte[BLOCK_SIZE];
+ System.arraycopy(data, i * BLOCK_SIZE, expected, 0, BLOCK_SIZE);
+ assertArrayEquals(expected, futures.get(i).get());
+ clones[i].close();
+ }
+ executor.shutdown();
+ }
+ }
+
+ @Test
+ public void testCloneAfterOriginalClosed() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE, 'i', 26);
+ Path path = createFile("testfile_clone_after_close", data);
+
+ FSDataInputStream cloned;
+ try (FSDataInputStream original = fs.open(path)) {
+ cloned = fs.cloneDataInputStream(original);
+ }
+ ContractTestUtils.verifyRead(cloned, data, 0, data.length);
+ cloned.close();
+ }
+
+ @Test
+ public void testCloneDataInputStreamEncrypted() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 2, 'j', 26);
+ Path path = new Path(ezPath, "encrypted_file");
+ ContractTestUtils.createFile(fs, path, true, data);
+
+ try (FSDataInputStream original = fs.open(path)) {
+ assertInstanceOf(CryptoInputStream.class, original.getWrappedStream(),
+ "Expected CryptoInputStream for encrypted file");
+
+ long countBefore = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+
+ try (FSDataInputStream cloned = fs.cloneDataInputStream(original)) {
+ long countAfter = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+ assertEquals(countBefore, countAfter,
+ "Cloning encrypted stream should not trigger "
+ + "GetBlockLocations RPC");
+
+ ContractTestUtils.verifyRead(cloned, data, 0, data.length);
+ }
+
+ ContractTestUtils.verifyRead(original, data, 0, data.length);
+ }
+ }
+
+ @Test
+ public void testOpenFileWithStatusEncrypted() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 2, 'k', 26);
+ Path path = new Path(ezPath, "encrypted_file2");
+ ContractTestUtils.createFile(fs, path, true, data);
+
+ RemoteIterator<LocatedFileStatus> iter = fs.listLocatedStatus(path);
+ HdfsLocatedFileStatus locatedStatus =
+ (HdfsLocatedFileStatus) iter.next();
+
+ long countBefore = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+
+ try (FSDataInputStream in = fs.openFile(path)
+ .withFileStatus(locatedStatus)
+ .build()
+ .get()) {
+ ContractTestUtils.verifyRead(in, data, 0, data.length);
+ }
+
+ long countAfter = getLong(getMetrics(NN_METRICS),
+ "GetBlockLocations");
+ assertEquals(countBefore, countAfter,
+ "Opening encrypted file with pre-fetched locations should not "
+ + "trigger GetBlockLocations RPC");
+ }
+
+ @Test
+ public void testCloneEncryptedIndependentPosition() throws Exception {
+ byte[] data = ContractTestUtils.dataset(BLOCK_SIZE * 2, 'l', 26);
+ Path path = new Path(ezPath, "encrypted_file3");
+ ContractTestUtils.createFile(fs, path, true, data);
+
+ try (FSDataInputStream original = fs.open(path)) {
+ byte[] buf1 = new byte[BLOCK_SIZE];
+ original.readFully(buf1);
+ assertEquals(BLOCK_SIZE, original.getPos());
+
+ try (FSDataInputStream cloned = fs.cloneDataInputStream(original)) {
+ assertEquals(0, cloned.getPos());
+
+ ContractTestUtils.verifyRead(cloned, data, 0, data.length);
+ }
+ }
+ }
+
+ private static long getLong(
+ org.apache.hadoop.metrics2.MetricsRecordBuilder rb, String name) {
+ return org.apache.hadoop.test.MetricsAsserts.getLongCounter(name, rb);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
