From 718b0a65362203d55b2451fe7f52d1a67626208f Mon Sep 17 00:00:00 2001 From: Ranga Reddy Date: Mon, 3 Aug 2026 12:36:35 +0530 Subject: [PATCH 1/3] 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. --- .../table/marker/WriteMarkersFactory.java | 2 +- .../apache/hudi/hadoop/fs/HadoopFSUtils.java | 27 ++++++++- .../fs/HoodieRetryWrapperFileSystem.java | 2 +- .../hadoop/fs/HoodieWrapperFileSystem.java | 2 +- .../storage/hadoop/HoodieHadoopStorage.java | 2 +- .../hudi/hadoop/fs/TestHadoopFSUtils.java | 56 +++++++++++++++++++ 6 files changed, 84 insertions(+), 7 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 2765ffdd6286a..8191e6d04cba4 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 static WriteMarkers get(MarkerType markerType, HoodieTable table, String } 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 e167657803143..f74da8c133ce3 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 @@ -277,7 +277,7 @@ private static FSDataInputStream getFSDataInputStreamForGCS(FSDataInputStream fs * @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 +285,28 @@ public static boolean isGCSFileSystem(FileSystem fs) { * 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()}. + * + *

{@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 it, and its scheme is what + * {@code getScheme()} returns wherever both are present. + * + * @param fs instance of {@link FileSystem} in use. + * @return the scheme of {@code fs}, or null if its URI carries none. + */ + public static String getScheme(FileSystem fs) { + try { + return fs.getScheme(); + } catch (UnsupportedOperationException e) { + return fs.getUri().getScheme(); + } } private static StorageConfiguration getStorageConf(Configuration conf, boolean copy) { @@ -294,7 +315,7 @@ private static StorageConfiguration getStorageConf(Configuration 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 c9d8fff3fbbd7..d7c1ca5f72fc8 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 Configuration getConf() { @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 24674ee725a90..01b7cb0f44ac8 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 @@ -160,7 +160,7 @@ public HoodieWrapperFileSystem(FileSystem fileSystem, ConsistencyGuard consisten public static Path convertToHoodiePath(StoragePath file, Configuration conf) { try { - String scheme = HadoopFSUtils.getFs(file.toString(), conf).getScheme(); + String scheme = HadoopFSUtils.getScheme(HadoopFSUtils.getFs(file.toString(), conf)); return convertPathWithScheme(convertToHadoopPath(file), getHoodieScheme(scheme)); } catch (HoodieIOException e) { throw e; 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 87a9ad1019f63..44d79cd14f3cf 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 @@ -111,7 +111,7 @@ public HoodieStorage newInstance(StoragePath path, StorageConfiguration stora @Override public String getScheme() { - return fs.getScheme(); + return HadoopFSUtils.getScheme(fs); } @Override 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 7768ff4feae7f..807af8510fbdb 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 @@ -22,22 +22,78 @@ import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathInfo; +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.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.File; +import java.io.IOException; +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; /** * 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. + * + *

{@link FilterFileSystem} has the same shape: it leaves {@code getScheme()} to the throwing base + * implementation while overriding {@code getUri()}. + */ + @Test + public void testGetFSDataInputStreamWhenGetSchemeIsUnimplemented(@TempDir File tempDir) throws IOException { + File file = new File(tempDir, "log.file"); + byte[] contents = new byte[] {1, 2, 3, 4}; + Files.write(file.toPath(), 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 = new FilterFileSystem(FileSystem.newInstanceLocal(new Configuration()))) { + // The premise: this is the call the read path used to make unguarded. + assertThrows(UnsupportedOperationException.class, fs::getScheme); + + 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), + "An implementation that overrides getScheme() should still be used directly"); + + try (FileSystem noScheme = new FilterFileSystem(localFs)) { + assertThrows(UnsupportedOperationException.class, noScheme::getScheme); + assertEquals("file", HadoopFSUtils.getScheme(noScheme), + "The scheme should come from getUri() when getScheme() is unimplemented"); + } + } + } + @ParameterizedTest @ValueSource(strings = { "/a/b/c", From c6e63473ab057d64677d27edec09e10db305ad23 Mon Sep 17 00:00:00 2001 From: Ranga Reddy Date: Mon, 3 Aug 2026 14:40:50 +0530 Subject: [PATCH 2/3] 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(). --- .../java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 807af8510fbdb..c142ca5b16353 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 @@ -84,12 +84,14 @@ public void testGetFSDataInputStreamWhenGetSchemeIsUnimplemented(@TempDir File t public void testGetSchemeFallsBackToTheUriWhenUnimplemented() throws IOException { try (FileSystem localFs = FileSystem.newInstanceLocal(new Configuration())) { assertEquals("file", HadoopFSUtils.getScheme(localFs), - "An implementation that overrides getScheme() should still be used directly"); + "LocalFileSystem overrides getScheme(), so the helper should return what it reports " + + "rather than falling back to getUri()"); try (FileSystem noScheme = new FilterFileSystem(localFs)) { assertThrows(UnsupportedOperationException.class, noScheme::getScheme); assertEquals("file", HadoopFSUtils.getScheme(noScheme), - "The scheme should come from getUri() when getScheme() is unimplemented"); + "FilterFileSystem does not override getScheme(), so the helper should fall back to " + + "getUri().getScheme()"); } } } From 5f96026bda7cbdeec1cc13f1ccdec7400e752b1d Mon Sep 17 00:00:00 2001 From: Ranga Reddy Date: Mon, 3 Aug 2026 19:13:15 +0530 Subject: [PATCH 3/3] 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. --- .../apache/hudi/hadoop/fs/HadoopFSUtils.java | 23 ++- .../hadoop/fs/HoodieWrapperFileSystem.java | 8 +- .../storage/hadoop/HoodieHadoopStorage.java | 12 ++ .../fs/TestFSUtilsWithRetryWrapperEnable.java | 16 +-- .../hudi/hadoop/fs/TestHadoopFSUtils.java | 133 ++++++++++++++++-- 5 files changed, 159 insertions(+), 33 deletions(-) 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 f74da8c133ce3..8c0293dbe7957 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.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; @@ -295,17 +296,31 @@ public static boolean isCHDFileSystem(FileSystem fs) { * {@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 it, and its scheme is what - * {@code getScheme()} returns wherever both are present. + * {@link FileSystem#getUri()} is abstract, so every implementation supplies one to fall back on. + * + *

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}, or null if its URI carries none. + * @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) { - return fs.getUri().getScheme(); + 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; } } 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 01b7cb0f44ac8..f8d731fd0ce21 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 HoodieWrapperFileSystem(FileSystem fileSystem, ConsistencyGuard consisten } public static Path convertToHoodiePath(StoragePath file, Configuration conf) { - try { - String scheme = HadoopFSUtils.getScheme(HadoopFSUtils.getFs(file.toString(), conf)); - 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 44d79cd14f3cf..067f77d491bd6 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 @@ */ 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 scheme = Lazy.lazily(this::resolveScheme); public HoodieHadoopStorage(StoragePath path, StorageConfiguration conf) { super(conf); @@ -111,6 +119,10 @@ public HoodieStorage newInstance(StoragePath path, StorageConfiguration stora @Override public String getScheme() { + return scheme.get(); + } + + private String resolveScheme() { return HadoopFSUtils.getScheme(fs); } 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 bb0b3608c7fc8..aaaf749e85d46 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.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 void testGetSchema() { 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 Configuration getConf() { 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 c142ca5b16353..da4e9a7500f42 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,14 +19,17 @@ 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; @@ -34,8 +37,8 @@ import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; -import java.io.File; import java.io.IOException; +import java.net.URI; import java.nio.file.Files; import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopFileStatus; @@ -43,8 +46,11 @@ 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} @@ -61,18 +67,15 @@ public class TestHadoopFSUtils { * implementation while overriding {@code getUri()}. */ @Test - public void testGetFSDataInputStreamWhenGetSchemeIsUnimplemented(@TempDir File tempDir) throws IOException { - File file = new File(tempDir, "log.file"); + 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.toPath(), contents); + 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 = new FilterFileSystem(FileSystem.newInstanceLocal(new Configuration()))) { - // The premise: this is the call the read path used to make unguarded. - assertThrows(UnsupportedOperationException.class, fs::getScheme); - + try (FileSystem fs = newFsWithoutGetScheme(FileSystem.newInstanceLocal(new Configuration()))) { try (FSDataInputStream stream = - HadoopFSUtils.getFSDataInputStream(fs, new StoragePath(file.toURI()), 1024, true)) { + 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()"); @@ -87,15 +90,117 @@ public void testGetSchemeFallsBackToTheUriWhenUnimplemented() throws IOException "LocalFileSystem overrides getScheme(), so the helper should return what it reports " + "rather than falling back to getUri()"); - try (FileSystem noScheme = new FilterFileSystem(localFs)) { - assertThrows(UnsupportedOperationException.class, noScheme::getScheme); - assertEquals("file", HadoopFSUtils.getScheme(noScheme), - "FilterFileSystem does not override getScheme(), so the helper should fall back to " - + "getUri().getScheme()"); + // 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..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",