This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 70a5a4dbd81c fix(fs): stop depending on the optional
FileSystem#getScheme() (#19470)
70a5a4dbd81c is described below
commit 70a5a4dbd81c0b73b32db577aa16dbadcca8ace6
Author: Ranga Reddy <[email protected]>
AuthorDate: Tue Aug 4 00:17:36 2026 +0530
fix(fs): stop depending on the optional FileSystem#getScheme() (#19470)
* fix(fs): stop depending on the optional FileSystem#getScheme()
FileSystem#getScheme() is optional in Hadoop: the base implementation throws
UnsupportedOperationException, and proxy implementations such as Presto's
PrestoS3FileSystem do not override it. Hudi called it unguarded on
filesystems it
did not implement, so opening a log file on such a filesystem failed with
"Not implemented by the PrestoS3FileSystem FileSystem implementation"
instead of
reading anything (HUDI-4602).
Adds HadoopFSUtils#getScheme(FileSystem), which returns fs.getScheme() and
falls
back to fs.getUri().getScheme() when it is unimplemented. getUri() is
abstract, so
every implementation supplies it, and its scheme is what getScheme() returns
wherever both are present. This is the same conclusion as #793, which
stopped
HoodieWrapperFileSystem calling getScheme() on the filesystem it wraps.
Routes the seven unguarded call sites through it: isGCSFileSystem and
isCHDFileSystem (the reported read path), registerFileSystem,
HoodieWrapperFileSystem#convertToHoodiePath,
HoodieRetryWrapperFileSystem#getScheme,
WriteMarkersFactory's HDFS gate, and HoodieHadoopStorage#getScheme, which
is what
the seven HoodieStorage#getScheme callers reach.
isGCSFileSystem's comparison is also flipped to put the constant first,
matching
isCHDFileSystem, so a filesystem whose URI carries no scheme returns false
rather
than throwing NullPointerException.
* test(fs): say which branch of the helper each assertion covers
Review nit: the assertion messages did not make clear what had gone wrong.
Each
now names the filesystem and the branch of the helper it pins -
LocalFileSystem
overriding getScheme() so the helper returns what it reports,
FilterFileSystem not
overriding it so the helper falls back to getUri().getScheme().
* fix(fs): fail loudly on an unresolvable scheme, and cover the sites this
reroutes
Review feedback, all of it well founded.
The fallback no longer returns null. InLineFileSystem is the
counter-example in this
module: getScheme() is "inlinefs" while getUri() is URI.create("inlinefs"),
which has no
colon and so no scheme, so the two are not interchangeable and the javadoc
claim that
they agree was simply wrong. A null surfaced far from the cause as "does
not support
scheme null" or "Unsupported scheme :null" with the
UnsupportedOperationException
discarded; it now throws with that exception chained. HoodieException
rather than
HoodieIOException, since the latter only accepts an IOException cause.
HoodieHadoopStorage memoizes the scheme. On a filesystem without
getScheme() the
fallback costs a thrown-and-caught exception, and this is called once per
log block via
StorageSchemes.isWriteTransactional and three times per immutable-file
write via
needCreateTempFile. A lazy field keeps all five constructors untouched.
Test coverage for what this actually reroutes, none of which any test
reached:
- registerFileSystem, HoodieWrapperFileSystem#convertToHoodiePath (the
write path) and
HoodieHadoopStorage#getScheme, via a LocalFileSystem subclass whose
getScheme() throws,
registered as fs.file.impl so it is reached through FileSystem.get.
- isGCSFileSystem and isCHDFileSystem, which become reachable for proxy
filesystems for
the first time here and select different stream wrappers: a scheme-less
filesystem
reporting gs:// now yields SchemeAwareFSDataInputStream and ofs:// yields
BoundedFsDataInputStream.
- the new unresolvable-scheme failure.
TestFSUtilsWithRetryWrapperEnable#testGetSchema has been inert since
HUDI-5286 added it:
it asserted on HoodieWrapperFileSystem#getScheme, which is uri.getScheme()
and never
dispatches into the retry wrapper, and FakeRemoteFileSystem overrode
getScheme() to
delegate to a real LocalFileSystem so it could not throw. Dropping that
override gives
the fake the PrestoS3FileSystem shape and the assertion now targets the
retry wrapper,
so it guards both HUDI-5286 and this change. Verified: it fails with the
pre-PR helper.
Also drops the try/catch in convertToHoodiePath that only rethrew
HoodieIOException
unchanged, dead since ef70de2bba7b, and the duplicated fixture and
redundant nested
close in TestHadoopFSUtils.
---
.../hudi/table/marker/WriteMarkersFactory.java | 2 +-
.../org/apache/hudi/hadoop/fs/HadoopFSUtils.java | 42 +++++-
.../hadoop/fs/HoodieRetryWrapperFileSystem.java | 2 +-
.../hudi/hadoop/fs/HoodieWrapperFileSystem.java | 8 +-
.../hudi/storage/hadoop/HoodieHadoopStorage.java | 14 +-
.../fs/TestFSUtilsWithRetryWrapperEnable.java | 16 +-
.../apache/hudi/hadoop/fs/TestHadoopFSUtils.java | 163 +++++++++++++++++++++
7 files changed, 226 insertions(+), 21 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java
index 2765ffdd6286..8191e6d04cba 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java
@@ -53,7 +53,7 @@ public class WriteMarkersFactory {
}
String basePath = table.getMetaClient().getBasePath().toString();
if (StorageSchemes.HDFS.getScheme().equals(
- HadoopFSUtils.getFs(basePath, table.getContext().getStorageConf(),
true).getScheme())) {
+ HadoopFSUtils.getScheme(HadoopFSUtils.getFs(basePath,
table.getContext().getStorageConf(), true)))) {
log.warn("Timeline-server-based markers are not supported for HDFS: "
+ "base path {}. Falling back to direct markers.", basePath);
return getDirectWriteMarkers(table, instantTime);
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
index e16765780314..8c0293dbe795 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
@@ -26,6 +26,7 @@ import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
@@ -277,7 +278,7 @@ public class HadoopFSUtils {
* @return true if the inputstream or the wrapped one is of type
GoogleHadoopFSInputStream
*/
public static boolean isGCSFileSystem(FileSystem fs) {
- return fs.getScheme().equals(StorageSchemes.GCS.getScheme());
+ return StorageSchemes.GCS.getScheme().equals(getScheme(fs));
}
/**
@@ -285,7 +286,42 @@ public class HadoopFSUtils {
* Wrapped by {@code BoundedFsDataInputStream}, to check whether the desired
offset is out of the file size in advance.
*/
public static boolean isCHDFileSystem(FileSystem fs) {
- return StorageSchemes.CHDFS.getScheme().equals(fs.getScheme());
+ return StorageSchemes.CHDFS.getScheme().equals(getScheme(fs));
+ }
+
+ /**
+ * Resolves the scheme of {@code fs} without depending on {@link
FileSystem#getScheme()}.
+ *
+ * <p>{@code getScheme()} is optional in Hadoop: {@link FileSystem}'s own
implementation throws
+ * {@link UnsupportedOperationException}, and proxy implementations such as
Presto's
+ * {@code PrestoS3FileSystem} do not override it, so calling it unguarded
turns an unrelated read into
+ * "Not implemented by the PrestoS3FileSystem FileSystem implementation"
(HUDI-4602).
+ * {@link FileSystem#getUri()} is abstract, so every implementation supplies
one to fall back on.
+ *
+ * <p>The two are not interchangeable, which is why {@code getScheme()} is
tried first:
+ * {@code InLineFileSystem} returns {@code "inlinefs"} from {@code
getScheme()} while its
+ * {@code getUri()} is {@code URI.create("inlinefs")}, which has no colon
and so carries no scheme at all.
+ * A URI with no scheme is therefore a resolution failure rather than a
value to pass on - returning null
+ * would surface much later as {@code does not support scheme null} or
{@code Unsupported scheme :null},
+ * with the original {@code UnsupportedOperationException} discarded.
+ *
+ * @param fs instance of {@link FileSystem} in use.
+ * @return the scheme of {@code fs}, never null.
+ * @throws HoodieException if {@code getScheme()} is unimplemented and the
URI carries no scheme.
+ */
+ public static String getScheme(FileSystem fs) {
+ try {
+ return fs.getScheme();
+ } catch (UnsupportedOperationException e) {
+ String scheme = fs.getUri().getScheme();
+ if (scheme == null) {
+ // HoodieException rather than HoodieIOException: the latter only
accepts an IOException cause, and
+ // discarding the UnsupportedOperationException is the thing being
fixed here.
+ throw new HoodieException("Cannot resolve the scheme of " +
fs.getClass().getName()
+ + ": getScheme() is unimplemented and its URI " + fs.getUri() + "
carries no scheme", e);
+ }
+ return scheme;
+ }
}
private static StorageConfiguration<Configuration>
getStorageConf(Configuration conf, boolean copy) {
@@ -294,7 +330,7 @@ public class HadoopFSUtils {
public static Configuration registerFileSystem(StoragePath file,
Configuration conf) {
Configuration returnConf = new Configuration(conf);
- String scheme = HadoopFSUtils.getFs(file.toString(), conf).getScheme();
+ String scheme = getScheme(HadoopFSUtils.getFs(file.toString(), conf));
returnConf.set("fs." + HoodieWrapperFileSystem.getHoodieScheme(scheme) +
".impl",
HoodieWrapperFileSystem.class.getName());
return returnConf;
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieRetryWrapperFileSystem.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieRetryWrapperFileSystem.java
index c9d8fff3fbbd..d7c1ca5f72fc 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieRetryWrapperFileSystem.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieRetryWrapperFileSystem.java
@@ -277,7 +277,7 @@ public class HoodieRetryWrapperFileSystem extends
FileSystem {
@Override
public String getScheme() {
- return fileSystem.getScheme();
+ return HadoopFSUtils.getScheme(fileSystem);
}
@Override
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java
index 24674ee725a9..f8d731fd0ce2 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java
@@ -159,12 +159,8 @@ public class HoodieWrapperFileSystem extends FileSystem {
}
public static Path convertToHoodiePath(StoragePath file, Configuration conf)
{
- try {
- String scheme = HadoopFSUtils.getFs(file.toString(), conf).getScheme();
- return convertPathWithScheme(convertToHadoopPath(file),
getHoodieScheme(scheme));
- } catch (HoodieIOException e) {
- throw e;
- }
+ String scheme =
HadoopFSUtils.getScheme(HadoopFSUtils.getFs(file.toString(), conf));
+ return convertPathWithScheme(convertToHadoopPath(file),
getHoodieScheme(scheme));
}
public static Path convertPathWithScheme(Path oldPath, String newScheme) {
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
index 87a9ad1019f6..067f77d491bd 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
@@ -20,6 +20,7 @@
package org.apache.hudi.storage.hadoop;
import org.apache.hudi.common.fs.ConsistencyGuard;
+import org.apache.hudi.common.util.Lazy;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieRetryWrapperFileSystem;
@@ -58,6 +59,13 @@ import static org.apache.hudi.hadoop.fs.HadoopFSUtils.getFs;
*/
public class HoodieHadoopStorage extends HoodieStorage {
private final FileSystem fs;
+ /**
+ * Resolved once. On a filesystem that does not implement {@code
getScheme()} the fallback in
+ * {@link HadoopFSUtils#getScheme} costs a thrown-and-caught exception, and
this is called once per log
+ * block via {@code StorageSchemes.isWriteTransactional} and three times per
immutable-file write via
+ * {@code needCreateTempFile}. {@code fs} is final, so the answer cannot
change.
+ */
+ private final Lazy<String> scheme = Lazy.lazily(this::resolveScheme);
public HoodieHadoopStorage(StoragePath path, StorageConfiguration<?> conf) {
super(conf);
@@ -111,7 +119,11 @@ public class HoodieHadoopStorage extends HoodieStorage {
@Override
public String getScheme() {
- return fs.getScheme();
+ return scheme.get();
+ }
+
+ private String resolveScheme() {
+ return HadoopFSUtils.getScheme(fs);
}
@Override
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
index bb0b3608c7fc..aaaf749e85d4 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
@@ -45,7 +45,6 @@ import java.net.URI;
import java.util.Arrays;
import java.util.List;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -107,9 +106,13 @@ public class TestFSUtilsWithRetryWrapperEnable extends
TestFSUtils {
FileSystem fileSystem =
new HoodieRetryWrapperFileSystem(fakeFs, maxRetryIntervalMs,
maxRetryNumbers,
initialRetryIntervalMs, "");
- HoodieWrapperFileSystem fs =
- new HoodieWrapperFileSystem(fileSystem, new NoOpConsistencyGuard());
- assertDoesNotThrow(fs::getScheme, "Method #getSchema does not implement
correctly");
+ // FakeRemoteFileSystem deliberately does not override getScheme(), so
FileSystem's own implementation
+ // throws - the PrestoS3FileSystem shape (HUDI-4602). Assert on the retry
wrapper itself: asserting on
+ // HoodieWrapperFileSystem instead would only exercise its own
uri.getScheme() and never reach here,
+ // which is why this guard was inert from the day HUDI-5286 added it.
+ assertThrows(UnsupportedOperationException.class, fakeFs::getScheme);
+ assertEquals("file", ((HoodieRetryWrapperFileSystem)
fileSystem).getScheme(),
+ "the retry wrapper should resolve the scheme of a filesystem that does
not implement getScheme()");
}
@Test
@@ -254,11 +257,6 @@ public class TestFSUtilsWithRetryWrapperEnable extends
TestFSUtils {
return fs.getConf();
}
- @Override
- public String getScheme() {
- return fs.getScheme();
- }
-
@Override
public short getDefaultReplication(Path path) {
return defaultReplication;
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
index 7768ff4feae7..da4e9a7500f4 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
@@ -19,25 +19,188 @@
package org.apache.hudi.hadoop.fs;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
+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.FilterFileSystem;
+import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+
import static
org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopFileStatus;
import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopPath;
import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath;
import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePathInfo;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Tests {@link HadoopFSUtils}
*/
public class TestHadoopFSUtils {
+ /**
+ * HUDI-4602: {@link FileSystem#getScheme()} is optional in Hadoop -- the
base implementation throws
+ * {@link UnsupportedOperationException} -- and proxy implementations such
as Presto's
+ * {@code PrestoS3FileSystem} do not override it. Opening a log file went
straight through
+ * {@code isGCSFileSystem}, so a MOR {@code _rt} query on Presto failed with
+ * "Not implemented by the PrestoS3FileSystem FileSystem implementation"
rather than reading anything.
+ *
+ * <p>{@link FilterFileSystem} has the same shape: it leaves {@code
getScheme()} to the throwing base
+ * implementation while overriding {@code getUri()}.
+ */
+ @Test
+ public void testGetFSDataInputStreamWhenGetSchemeIsUnimplemented(@TempDir
java.nio.file.Path tempDir) throws IOException {
+ java.nio.file.Path file = tempDir.resolve("log.file");
+ byte[] contents = new byte[] {1, 2, 3, 4};
+ Files.write(file, contents);
+ // newInstanceLocal rather than getLocal, so closing this does not evict a
cached FileSystem that
+ // other tests in the same JVM share.
+ try (FileSystem fs = newFsWithoutGetScheme(FileSystem.newInstanceLocal(new
Configuration()))) {
+ try (FSDataInputStream stream =
+ HadoopFSUtils.getFSDataInputStream(fs, new
StoragePath(file.toUri()), 1024, true)) {
+ byte[] read = new byte[contents.length];
+ stream.readFully(read);
+ assertArrayEquals(contents, read, "The read path should not depend on
the optional getScheme()");
+ }
+ }
+ }
+
+ @Test
+ public void testGetSchemeFallsBackToTheUriWhenUnimplemented() throws
IOException {
+ try (FileSystem localFs = FileSystem.newInstanceLocal(new
Configuration())) {
+ assertEquals("file", HadoopFSUtils.getScheme(localFs),
+ "LocalFileSystem overrides getScheme(), so the helper should return
what it reports "
+ + "rather than falling back to getUri()");
+
+ // FilterFileSystem#close closes the delegate, so the wrapper is not
given its own block: it owns
+ // nothing, and closing it here would close localFs a second time.
+ FileSystem noScheme = newFsWithoutGetScheme(localFs);
+ assertEquals("file", HadoopFSUtils.getScheme(noScheme),
+ "FilterFileSystem does not override getScheme(), so the helper
should fall back to "
+ + "getUri().getScheme()");
+ }
+ }
+
+ /**
+ * A URI with no scheme cannot stand in for an unimplemented {@code
getScheme()}. {@code InLineFileSystem}
+ * is the case in this module: {@code getScheme()} returns "inlinefs" while
{@code getUri()} is
+ * {@code URI.create("inlinefs")}, which has no colon and so no scheme.
Returning null there would surface
+ * far away as "does not support scheme null" with the original failure
discarded, so it must fail here.
+ */
+ @Test
+ public void testGetSchemeFailsLoudlyWhenNeitherSourceHasOne() throws
IOException {
+ try (FileSystem localFs = FileSystem.newInstanceLocal(new
Configuration())) {
+ FileSystem schemeless = new NoSchemeFileSystem(localFs,
URI.create("inlinefs"));
+
+ HoodieException thrown =
+ assertThrows(HoodieException.class, () ->
HadoopFSUtils.getScheme(schemeless));
+ assertTrue(thrown.getMessage().contains("carries no scheme"),
+ () -> "the failure should say the URI carries no scheme, but was: "
+ thrown.getMessage());
+ assertInstanceOf(UnsupportedOperationException.class, thrown.getCause(),
+ "the original getScheme() failure must be chained rather than
discarded");
+ }
+ }
+
+ /**
+ * The three call sites this rerouted that no test in the repo reached:
{@code registerFileSystem},
+ * {@code HoodieWrapperFileSystem#convertToHoodiePath} - which is on the
write path, via
+ * {@code HoodieBaseParquetWriter} and friends - and {@code
HoodieHadoopStorage#getScheme}. All three threw
+ * {@link UnsupportedOperationException} on a filesystem without {@code
getScheme()} before this change.
+ */
+ @Test
+ public void testCallSitesWorkOnAFileSystemWithoutGetScheme(@TempDir
java.nio.file.Path tempDir) {
+ Configuration conf = new Configuration();
+ conf.setClass("fs.file.impl", NoSchemeLocalFileSystem.class,
FileSystem.class);
+ StoragePath path = new StoragePath(tempDir.toUri());
+
+ assertDoesNotThrow(() -> HadoopFSUtils.registerFileSystem(path, conf),
+ "registerFileSystem resolves the scheme to build the fs.<scheme>.impl
key");
+ assertDoesNotThrow(() -> HoodieWrapperFileSystem.convertToHoodiePath(path,
conf),
+ "convertToHoodiePath is on the write path and resolves the scheme to
rewrite it");
+ assertEquals("file", new HoodieHadoopStorage(path,
HadoopFSUtils.getStorageConf(conf)).getScheme(),
+ "HoodieHadoopStorage#getScheme is what HoodieStorage callers reach");
+ }
+
+ /**
+ * {@code isGCSFileSystem} and {@code isCHDFileSystem} become reachable for
a filesystem without
+ * {@code getScheme()} for the first time with this change, and they select
different stream wrappers.
+ * Neither predicate had a test before.
+ */
+ @ParameterizedTest
+ @CsvSource({
+ "gs://bucket, org.apache.hudi.hadoop.fs.SchemeAwareFSDataInputStream",
+ "ofs://cluster, org.apache.hudi.hadoop.fs.BoundedFsDataInputStream"
+ })
+ public void testSchemeSpecificStreamIsSelectedWithoutGetScheme(String uri,
String expectedStream,
+ @TempDir
java.nio.file.Path tempDir) throws IOException {
+ java.nio.file.Path file = tempDir.resolve("log.file");
+ Files.write(file, new byte[] {1, 2, 3, 4});
+ try (FileSystem localFs = FileSystem.newInstanceLocal(new
Configuration())) {
+ // Reports a gs:// or ofs:// URI while leaving getScheme() to the
throwing base implementation.
+ FileSystem fs = new NoSchemeFileSystem(localFs, URI.create(uri));
+ assertThrows(UnsupportedOperationException.class, fs::getScheme);
+
+ try (FSDataInputStream stream =
+ HadoopFSUtils.getFSDataInputStream(fs, new
StoragePath(file.toUri()), 1024, true)) {
+ assertEquals(expectedStream, stream.getClass().getName(),
+ "the scheme-specific wrapper should be selected from the
fallback-resolved scheme");
+ }
+ }
+ }
+
+ /** A FileSystem with the reported shape: {@code getUri()} works, {@code
getScheme()} throws. */
+ private static FileSystem newFsWithoutGetScheme(FileSystem delegate) {
+ FileSystem fs = new FilterFileSystem(delegate);
+ // The premise of every assertion below: this is the call the read path
used to make unguarded.
+ assertThrows(UnsupportedOperationException.class, fs::getScheme);
+ return fs;
+ }
+
+ /** Same shape, but reporting a URI of our choosing so scheme-specific
branches can be reached. */
+ private static class NoSchemeFileSystem extends FilterFileSystem {
+ private final URI uri;
+
+ NoSchemeFileSystem(FileSystem delegate, URI uri) {
+ super(delegate);
+ this.uri = uri;
+ }
+
+ @Override
+ public URI getUri() {
+ return uri;
+ }
+ }
+
+ /**
+ * A {@link LocalFileSystem} that does not implement {@code getScheme()}, so
it can be registered as
+ * {@code fs.file.impl} and reached through the normal {@code
FileSystem.get} path.
+ */
+ public static class NoSchemeLocalFileSystem extends LocalFileSystem {
+ @Override
+ public String getScheme() {
+ throw new UnsupportedOperationException(
+ "Not implemented by the NoSchemeLocalFileSystem FileSystem
implementation");
+ }
+ }
+
@ParameterizedTest
@ValueSource(strings = {
"/a/b/c",
