diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java index 8ec43d9781d4b..4d295ad9b50b6 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java @@ -18,7 +18,6 @@ package org.apache.hudi.client; -import org.apache.hudi.client.common.HoodieFlinkEngineContext; import org.apache.hudi.client.embedded.EmbeddedTimelineService; import org.apache.hudi.common.engine.HoodieEngineContext; import org.apache.hudi.common.model.HoodieCommitMetadata; @@ -37,7 +36,6 @@ import org.apache.hudi.exception.HoodieCommitException; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; -import org.apache.hudi.hadoop.fs.HadoopFSUtils; import org.apache.hudi.metadata.FlinkHoodieBackedTableMetadataWriter; import org.apache.hudi.metadata.HoodieBackedTableMetadataWriter; import org.apache.hudi.storage.StorageConfiguration; @@ -46,7 +44,6 @@ import org.apache.hudi.table.action.HoodieWriteMetadata; import org.apache.hudi.table.action.compact.CompactHelpers; import org.apache.hudi.table.marker.WriteMarkersFactory; -import org.apache.hudi.util.FlinkClientUtil; import lombok.extern.slf4j.Slf4j; @@ -187,7 +184,7 @@ protected HoodieTable createTable(HoodieWriteConfig config, StorageConfiguration */ private HoodieBackedTableMetadataWriter initMetadataWriter(Option latestPendingInstant) { return (HoodieBackedTableMetadataWriter) FlinkHoodieBackedTableMetadataWriter.create( - HadoopFSUtils.getStorageConf(FlinkClientUtil.getHadoopConf()), this.config, HoodieFlinkEngineContext.DEFAULT, latestPendingInstant); + storageConf, this.config, context, latestPendingInstant); } public void initMetadataTable() { diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/FlinkClientUtil.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/FlinkClientUtil.java index d4e67507b6ae5..baa8fe42de7ae 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/FlinkClientUtil.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/FlinkClientUtil.java @@ -24,6 +24,7 @@ import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.util.StringUtils; import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieValidationException; import org.apache.hudi.hadoop.fs.HadoopFSUtils; import org.apache.flink.api.java.hadoop.mapred.utils.HadoopUtils; @@ -35,6 +36,7 @@ import org.slf4j.LoggerFactory; import java.io.File; +import java.util.Arrays; import java.util.Map; /** @@ -44,6 +46,7 @@ public class FlinkClientUtil { private static final Logger LOG = LoggerFactory.getLogger(FlinkClientUtil.class); public static String FLINK_HADOOP_CONFIG_PREFIX = "flink.hadoop"; public static String FLINK_CONF_DIR = "FLINK_CONF_DIR"; + private static final String[] HADOOP_SITE_FILES = {"core-site.xml", "hdfs-site.xml", "yarn-site.xml", "mapred-site.xml"}; /** * Creates the meta client. @@ -70,6 +73,19 @@ public static org.apache.hadoop.conf.Configuration getHadoopConf() { hadoopConf = new org.apache.hadoop.conf.Configuration(); } + return applyFlinkHadoopOptions(hadoopConf); + } + + /** + * Loads an explicit Hadoop configuration directory, then applies the Hadoop options from + * FLINK_CONF_DIR, as in environment-based discovery. Invalid explicit directories fail fast. + */ + public static org.apache.hadoop.conf.Configuration getHadoopConf(String hadoopConfDir) { + validateHadoopConfDir(hadoopConfDir); + return applyFlinkHadoopOptions(getHadoopConfiguration(hadoopConfDir)); + } + + private static org.apache.hadoop.conf.Configuration applyFlinkHadoopOptions(org.apache.hadoop.conf.Configuration hadoopConf) { String confDir = System.getenv(FLINK_CONF_DIR); if (!StringUtils.isNullOrEmpty(confDir)) { LOG.info("Flink conf dir: {}", confDir); @@ -86,35 +102,36 @@ public static org.apache.hadoop.conf.Configuration getHadoopConf() { return hadoopConf; } - /** - * Returns a new Hadoop Configuration object using the path to the hadoop conf configured. - * - * @param hadoopConfDir Hadoop conf directory path. - * @return A Hadoop configuration instance. - */ + private static void validateHadoopConfDir(String hadoopConfDir) { + if (hadoopConfDir == null || hadoopConfDir.trim().isEmpty()) { + throw new HoodieValidationException("'hadoop_conf.dir' must not be empty. Leave it unset to use environment discovery"); + } + File dir = new File(hadoopConfDir); + if (!dir.isDirectory()) { + throw new HoodieValidationException("Invalid 'hadoop_conf.dir': directory does not exist or is not a directory: " + hadoopConfDir); + } + boolean hasSiteFile = Arrays.stream(HADOOP_SITE_FILES) + .map(name -> new File(dir, name)) + .anyMatch(File::isFile); + if (!hasSiteFile) { + throw new HoodieValidationException("Invalid 'hadoop_conf.dir': no supported Hadoop site files " + + "(core-site.xml, hdfs-site.xml, yarn-site.xml, mapred-site.xml) found under directory: " + hadoopConfDir); + } + } + private static org.apache.hadoop.conf.Configuration getHadoopConfiguration(String hadoopConfDir) { - if (new File(hadoopConfDir).exists()) { - org.apache.hadoop.conf.Configuration hadoopConfiguration = new org.apache.hadoop.conf.Configuration(); - File coreSite = new File(hadoopConfDir, "core-site.xml"); - if (coreSite.exists()) { - hadoopConfiguration.addResource(new Path(coreSite.getAbsolutePath())); - } - File hdfsSite = new File(hadoopConfDir, "hdfs-site.xml"); - if (hdfsSite.exists()) { - hadoopConfiguration.addResource(new Path(hdfsSite.getAbsolutePath())); - } - File yarnSite = new File(hadoopConfDir, "yarn-site.xml"); - if (yarnSite.exists()) { - hadoopConfiguration.addResource(new Path(yarnSite.getAbsolutePath())); - } - // Add mapred-site.xml. We need to read configurations like compression codec. - File mapredSite = new File(hadoopConfDir, "mapred-site.xml"); - if (mapredSite.exists()) { - hadoopConfiguration.addResource(new Path(mapredSite.getAbsolutePath())); + File dir = new File(hadoopConfDir); + if (!dir.exists()) { + return null; + } + org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration(); + for (String name : HADOOP_SITE_FILES) { + File site = new File(dir, name); + if (site.exists()) { + hadoopConf.addResource(new Path(site.getAbsolutePath())); } - return hadoopConfiguration; } - return null; + return hadoopConf; } /** diff --git a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestHoodieFlinkTableServiceClient.java b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestHoodieFlinkTableServiceClient.java index 7ddacba98a661..7f9ffeacae47a 100644 --- a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestHoodieFlinkTableServiceClient.java +++ b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestHoodieFlinkTableServiceClient.java @@ -64,6 +64,8 @@ import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.ArgumentMatchers.same; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -112,7 +114,11 @@ void testInitMetadataTableRespectsStreamingWriteFlag(boolean metadataStreamingWr writerFactory.when(() -> FlinkHoodieBackedTableMetadataWriter.create(any(), any(), any(), any())) .thenReturn(metadataWriter); + context.getStorageConf().set("dfs.nameservices", "remote-cluster"); tableServiceClient.initMetadataTable(); + writerFactory.verify(() -> FlinkHoodieBackedTableMetadataWriter.create( + argThat(conf -> "remote-cluster".equals(conf.getString("dfs.nameservices").orElse(null))), + same(writeConfig), same(context), any())); } finally { tableServiceClient.close(); } diff --git a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestFlinkClientUtil.java b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestFlinkClientUtil.java index cc4df7c103e47..abdd5d6b050d4 100644 --- a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestFlinkClientUtil.java +++ b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestFlinkClientUtil.java @@ -21,9 +21,13 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.GlobalConfiguration; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import org.mockito.MockedStatic; import org.mockito.Mockito; +import java.nio.file.Files; +import java.nio.file.Path; + import static org.apache.hudi.util.FlinkClientUtil.FLINK_HADOOP_CONFIG_PREFIX; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.ArgumentMatchers.anyString; @@ -48,4 +52,22 @@ public void testGetHadoopConf() throws Exception { assertEquals(CRYPTO_RETRIEVER_CLASS, hadoopConf.get(CRYPTO_RETRIEVER_KEY)); } } + + @Test + public void testExplicitDirectoryPreservesFlinkHadoopOptions(@TempDir Path dir) throws Exception { + Files.writeString(dir.resolve("core-site.xml"), "fs.defaultFS" + + "hdfs://remote-cluster" + CRYPTO_RETRIEVER_KEY + + "site-value"); + try (MockedStatic global = Mockito.mockStatic(GlobalConfiguration.class)) { + Configuration flinkConf = new Configuration(); + flinkConf.setString(FLINK_CRYPTO_RETRIEVER_KEY, CRYPTO_RETRIEVER_CLASS); + flinkConf.setString(FLINK_HADOOP_CONFIG_PREFIX + "test.marker", "flink-only"); + global.when(() -> GlobalConfiguration.loadConfiguration(anyString())).thenReturn(flinkConf); + org.apache.hadoop.conf.Configuration hadoopConf = FlinkClientUtil.getHadoopConf(dir.toString()); + assertEquals("hdfs://remote-cluster", hadoopConf.get("fs.defaultFS")); + assertEquals(CRYPTO_RETRIEVER_CLASS, hadoopConf.get(CRYPTO_RETRIEVER_KEY)); + assertEquals("flink-only", hadoopConf.get("test.marker")); + } + } + } diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java index 6ec2adea71164..e0758e447731c 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java @@ -1535,6 +1535,17 @@ public class FlinkOptions extends HoodieConfig { + "Each task manager will create a unique subdirectory under this path. " + "The directory is cleaned up when the lookup function is closed."); + @AdvancedConfig + public static final ConfigOption HADOOP_CONF_DIR = ConfigOptions + .key("hadoop_conf.dir") + .stringType() + .noDefaultValue() + .withDescription("Local Hadoop configuration directory for cross-cluster reads and writes. " + + "It must contain at least one readable core-site.xml, hdfs-site.xml, yarn-site.xml or mapred-site.xml " + + "and be available at the same path on the client, JobManager and TaskManagers. " + + "Invalid explicit directories fail fast; when unset, environment discovery is used. " + + "Hadoop properties from FLINK_CONF_DIR override site files, and table-level hadoop.* options override both."); + // ------------------------------------------------------------------------- // Utilities // ------------------------------------------------------------------------- diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java index 12d920ac98705..0a10338e81834 100644 --- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java +++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java @@ -49,7 +49,9 @@ public static org.apache.hadoop.conf.Configuration getParquetConf( * along with some configurations necessary to construct file readers/writers. */ public static org.apache.hadoop.conf.Configuration getHadoopConf(Configuration conf) { - org.apache.hadoop.conf.Configuration hadoopConf = FlinkClientUtil.getHadoopConf(); + String hadoopConfDir = conf.getOptional(FlinkOptions.HADOOP_CONF_DIR).orElse(null); + org.apache.hadoop.conf.Configuration hadoopConf = hadoopConfDir == null + ? FlinkClientUtil.getHadoopConf() : FlinkClientUtil.getHadoopConf(hadoopConfDir); Map hadoopOptions = FlinkOptions.getPropertiesWithPrefix(conf.toMap(), HADOOP_PREFIX); hadoopOptions.forEach(hadoopConf::set); // kind of hacky: flink specific IO options. diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestHadoopConfigurations.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestHadoopConfigurations.java new file mode 100644 index 0000000000000..078c05cb0b620 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestHadoopConfigurations.java @@ -0,0 +1,272 @@ +/* + * 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.hudi.configuration; + +import org.apache.hudi.exception.HoodieValidationException; + +import org.apache.flink.configuration.Configuration; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Test cases for {@link HadoopConfigurations#getHadoopConf(Configuration)}, + * covering the {@code hadoop_conf.dir} option used for cross-cluster access. + */ +public class TestHadoopConfigurations { + + @TempDir + File tempDir; + + // ------------------------------------------------------------------------- + // Helpers + + private File createHadoopConfDir(String... siteFileContents) throws IOException { + File confDir = new File(tempDir, "hadoop-conf-" + System.nanoTime()); + assertTrue(confDir.mkdirs(), "Failed to create hadoop conf dir"); + for (String content : siteFileContents) { + // content format: "filename|xml-body" + String[] parts = content.split("\\|", 2); + writeSiteFile(new File(confDir, parts[0]), parts[1]); + } + return confDir; + } + + private void writeSiteFile(File file, String content) throws IOException { + try (FileWriter writer = new FileWriter(file)) { + writer.write(content); + } + } + + private static String siteXml(String name, String value) { + return "\n" + + "\n" + + " \n" + + " " + name + "\n" + + " " + value + "\n" + + " \n" + + "\n"; + } + + // ------------------------------------------------------------------------- + // Load explicit configuration directories + // ------------------------------------------------------------------------- + + @Test + public void testGetHadoopConfLoadsCoreSiteFromConfDir() throws Exception { + File confDir = createHadoopConfDir( + "core-site.xml|" + siteXml("fs.defaultFS", "hdfs://remote-cluster:8020")); + + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), confDir.getAbsolutePath()); + + org.apache.hadoop.conf.Configuration hadoopConf = + HadoopConfigurations.getHadoopConf(conf); + + assertEquals("hdfs://remote-cluster:8020", hadoopConf.get("fs.defaultFS"), + "fs.defaultFS should be loaded from the specified hadoop conf dir"); + } + + @Test + public void testGetHadoopConfLoadsHdfsSiteFromConfDir() throws Exception { + File confDir = createHadoopConfDir( + "hdfs-site.xml|" + siteXml("dfs.nameservices", "remote-ns")); + + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), confDir.getAbsolutePath()); + + org.apache.hadoop.conf.Configuration hadoopConf = + HadoopConfigurations.getHadoopConf(conf); + + assertEquals("remote-ns", hadoopConf.get("dfs.nameservices"), + "hdfs-site.xml properties should be loaded from the specified conf dir"); + } + + @Test + public void testGetHadoopConfMergesMultipleSiteFiles() throws Exception { + File confDir = createHadoopConfDir( + "core-site.xml|" + siteXml("fs.defaultFS", "hdfs://remote-cluster:8020"), + "hdfs-site.xml|" + siteXml("dfs.nameservices", "remote-ns"), + "yarn-site.xml|" + siteXml("yarn.resourcemanager.address", "remote-rm:8032")); + + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), confDir.getAbsolutePath()); + + org.apache.hadoop.conf.Configuration hadoopConf = + HadoopConfigurations.getHadoopConf(conf); + + assertEquals("hdfs://remote-cluster:8020", hadoopConf.get("fs.defaultFS")); + assertEquals("remote-ns", hadoopConf.get("dfs.nameservices")); + assertEquals("remote-rm:8032", hadoopConf.get("yarn.resourcemanager.address")); + } + + // ------------------------------------------------------------------------- + // Table-level Hadoop option precedence + // ------------------------------------------------------------------------- + + @Test + public void testIndividualHadoopOptionsOverrideConfDir() throws Exception { + File confDir = createHadoopConfDir( + "core-site.xml|" + siteXml("fs.defaultFS", "hdfs://remote-cluster:8020")); + + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), confDir.getAbsolutePath()); + // Individual hadoop.* option should take the highest precedence + conf.setString("hadoop.fs.defaultFS", "hdfs://override-cluster:8020"); + + org.apache.hadoop.conf.Configuration hadoopConf = + HadoopConfigurations.getHadoopConf(conf); + + assertEquals("hdfs://override-cluster:8020", hadoopConf.get("fs.defaultFS"), + "hadoop.* prefixed options should override values from hadoop_conf.dir"); + } + + @Test + public void testHadoopConfDirKeyNotLeakedIntoHadoopConf() throws Exception { + File confDir = createHadoopConfDir( + "core-site.xml|" + siteXml("fs.defaultFS", "hdfs://remote-cluster:8020")); + + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), confDir.getAbsolutePath()); + + org.apache.hadoop.conf.Configuration hadoopConf = + HadoopConfigurations.getHadoopConf(conf); + + // 'hadoop_conf.dir' uses underscore separator and never matches the 'hadoop.' + // prefix, so it must not be forwarded as a Hadoop property. + assertNull(hadoopConf.get("conf.dir"), + "hadoop_conf.dir should not leak into Hadoop conf as 'conf.dir'"); + assertNull(hadoopConf.get("hadoop_conf.dir"), + "hadoop_conf.dir is a connector option, not a Hadoop property"); + } + + // ------------------------------------------------------------------------- + // Reject invalid explicit configuration directories + // ------------------------------------------------------------------------- + + @Test + public void testGetHadoopConfWithNonExistentConfDir() { + String nonExistentDir = new File(tempDir, "does-not-exist").getAbsolutePath(); + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), nonExistentDir); + + HoodieValidationException e = assertThrows(HoodieValidationException.class, + () -> HadoopConfigurations.getHadoopConf(conf), + "Explicitly configured non-existent dir should fail fast"); + + assertTrue(e.getMessage().contains(nonExistentDir), + "Error message should contain the invalid conf dir path"); + assertTrue(e.getMessage().contains(FlinkOptions.HADOOP_CONF_DIR.key()), + "Error message should mention the option key"); + } + + @Test + public void testGetHadoopConfWithConfDirBeingAFile() throws Exception { + // Path exists but is a regular file, not a directory + File file = new File(tempDir, "not-a-dir"); + assertTrue(file.createNewFile()); + + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), file.getAbsolutePath()); + + HoodieValidationException e = assertThrows(HoodieValidationException.class, + () -> HadoopConfigurations.getHadoopConf(conf)); + + assertTrue(e.getMessage().contains(file.getAbsolutePath())); + } + + @Test + public void testGetHadoopConfWithConfDirMissingSiteFiles() { + // Directory exists but contains no *-site.xml — still a config error. + // An empty Configuration would silently fall back fs.defaultFS to local FS. + File emptyDir = new File(tempDir, "empty-conf-dir"); + assertTrue(emptyDir.mkdirs()); + + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), emptyDir.getAbsolutePath()); + + HoodieValidationException e = assertThrows(HoodieValidationException.class, + () -> HadoopConfigurations.getHadoopConf(conf)); + + assertTrue(e.getMessage().contains(emptyDir.getAbsolutePath())); + assertTrue(e.getMessage().contains("core-site.xml"), + "Error message should mention expected site files"); + } + + @Test + public void testGetHadoopConfWithEmptyStringConfDir() { + // Explicitly set to empty string is also a configuration error + Configuration conf = new Configuration(); + conf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), ""); + + assertThrows(HoodieValidationException.class, + () -> HadoopConfigurations.getHadoopConf(conf)); + } + + // Unconfigured options — environment discovery fallback + @Test + public void testGetHadoopConfWithoutConfDirFallsBackToEnvironment() { + // Without the option, getHadoopConf delegates to FlinkClientUtil + // environment discovery — must not throw, and hadoop.* overrides + // should still be applied on top. + Configuration conf = new Configuration(); + conf.setString("hadoop.test.marker", "marker-value"); + + org.apache.hadoop.conf.Configuration hadoopConf = + HadoopConfigurations.getHadoopConf(conf); + + assertNotNull(hadoopConf); + assertEquals("marker-value", hadoopConf.get("test.marker"), + "hadoop.* prefixed options should be forwarded even without hadoop_conf.dir"); + } + + @Test + public void testUnsupportedSiteFileDoesNotValidateDirectory() throws Exception { + File dir = createHadoopConfDir("hive-site.xml|" + siteXml("fs.defaultFS", "hdfs://ignored")); + Configuration conf = new Configuration(); + conf.set(FlinkOptions.HADOOP_CONF_DIR, dir.getAbsolutePath()); + assertThrows(HoodieValidationException.class, () -> HadoopConfigurations.getHadoopConf(conf)); + } + + @Test + public void testSiteFileMustBeRegularFile() throws Exception { + File dir = createHadoopConfDir(); + assertTrue(new File(dir, "core-site.xml").mkdir()); + Configuration conf = new Configuration(); + conf.set(FlinkOptions.HADOOP_CONF_DIR, dir.getAbsolutePath()); + assertThrows(HoodieValidationException.class, () -> HadoopConfigurations.getHadoopConf(conf)); + } + + @Test + public void testWhitespaceConfDirFails() { + Configuration conf = new Configuration(); + conf.set(FlinkOptions.HADOOP_CONF_DIR, " "); + assertThrows(HoodieValidationException.class, () -> HadoopConfigurations.getHadoopConf(conf)); + } + +} diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java index 5740a5e2e6d3a..c877fda22097c 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java @@ -37,6 +37,7 @@ import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.configuration.HadoopConfigurations; +import org.apache.hudi.hadoop.fs.NonLocalSchemeLocalFileSystem; import org.apache.hudi.index.HoodieIndex; import org.apache.hudi.index.bucket.partition.PartitionBucketIndexUtils; import org.apache.hudi.metadata.HoodieTableMetadata; @@ -86,6 +87,7 @@ import org.junit.jupiter.params.provider.ValueSource; import java.io.File; +import java.io.FileWriter; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; @@ -3670,6 +3672,156 @@ void testReadWithParquetPredicatePushDown() { + "+I[id8, Han, 56, 1970-01-01T00:00:08, par4]]"); } + // ------------------------------------------------------------------------- + // Cross-cluster access via hadoop_conf.dir + // ------------------------------------------------------------------------- + + /** + * Batch write then batch read, with 'hadoop_conf.dir' pointing at a conf + * directory that registers a non-local filesystem backed by local files. + */ + @Test + public void testWriteAndReadWithHadoopConfDir() throws Exception { + File hadoopConfDir = createHadoopConfDirWithRemoteFs(); + + StoragePath path = new StoragePath("viewfs://remote-cluster" + tempFile.getAbsolutePath()); + + // ---- write ---- + TableEnvironment writeEnv = batchTableEnv; + String writeDDL = sql("source") + .option(FlinkOptions.PATH, path) + .option(FlinkOptions.HADOOP_CONF_DIR, hadoopConfDir.getAbsolutePath()) + .option(FlinkOptions.METADATA_ENABLED, true) + .end(); + writeEnv.executeSql(writeDDL); + writeEnv.executeSql( + "insert into source values " + + "('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1')," + + "('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1')") + .await(); + + // ---- read ---- + String readDDL = sql("t1") + .option(FlinkOptions.PATH, path) + .option(FlinkOptions.HADOOP_CONF_DIR, hadoopConfDir.getAbsolutePath()) + .option(FlinkOptions.METADATA_ENABLED, true) + .end(); + batchTableEnv.executeSql(readDDL); + + List result = CollectionUtil.iterableToList( + () -> batchTableEnv.sqlQuery("select * from t1").execute().collect()); + assertRowsEquals(result, "[+I[id1, Danny, 23, 1970-01-01T00:00:01, par1], " + + "+I[id2, Stephen, 33, 1970-01-01T00:00:02, par1]]"); + } + + /** + * Incremental read with 'hadoop_conf.dir': the read path must use the + * configured conf dir (via HoodieTableSource / StreamReadMonitoringFunction), + * not the environment-discovered configuration. + */ + @Test + public void testIncrementalReadWithHadoopConfDir() throws Exception { + File hadoopConfDir = createHadoopConfDirWithRemoteFs(); + StoragePath path = new StoragePath("viewfs://remote-cluster" + tempFile.getAbsolutePath()); + Configuration conf = TestConfigurations.getDefaultConf(path.toString()); + conf.set(FlinkOptions.HADOOP_CONF_DIR, hadoopConfDir.getAbsolutePath()); + TestData.writeData(TestData.dataSetInsert(1, 2), conf); + TestData.writeData(TestData.dataSetInsert(3, 4), conf); + String secondCommit = StreamerUtil.createMetaClient(conf).getCommitsTimeline() + .filterCompletedInstants().lastInstant().get().requestedTime(); + + String ddl = sql("t1") + .option(FlinkOptions.PATH, path) + .option(FlinkOptions.QUERY_TYPE, FlinkOptions.QUERY_TYPE_INCREMENTAL) + .option(FlinkOptions.READ_START_COMMIT, secondCommit) + .option(FlinkOptions.HADOOP_CONF_DIR, hadoopConfDir.getAbsolutePath()) + .end(); + batchTableEnv.executeSql(ddl); + + List result = CollectionUtil.iterableToList( + () -> batchTableEnv.sqlQuery("select * from t1").execute().collect()); + assertRowsEquals(result, TestData.dataSetInsert(3, 4)); + } + + /** + * 'hadoop_conf.dir' pointing at a non-existent directory must fail fast — + * silently falling back to the environment configuration would resolve + * fs.defaultFS to the local filesystem instead of the intended cluster. + */ + @Test + public void testReadWithInvalidHadoopConfDirFails() { + StoragePath path = new StoragePath(tempFile.toURI()); + String nonExistentDir = new File(tempFile, "no-such-conf-dir").getAbsolutePath(); + + String ddl = sql("t1") + .option(FlinkOptions.PATH, path) + .option(FlinkOptions.HADOOP_CONF_DIR, nonExistentDir) + .end(); + + // The table source validates the conf dir when the catalog/planner + // instantiates it (DDL registration or first query planning). + batchTableEnv.executeSql(ddl); + Throwable e = assertThrows(Exception.class, + () -> batchTableEnv.sqlQuery("select * from t1").execute().collect().next()); + assertTrue(getRootMessage(e).contains(nonExistentDir) + && getRootMessage(e).contains(FlinkOptions.HADOOP_CONF_DIR.key()), + "Error should name the invalid conf dir and the option key, got: " + + getRootMessage(e)); + } + + /** + * An existing directory that contains no Hadoop site files is also a + * configuration error — an empty Configuration would again resolve to the + * local filesystem. + */ + @Test + public void testReadWithEmptyHadoopConfDirFails() { + StoragePath path = new StoragePath(tempFile.toURI()); + File emptyDir = new File(tempFile, "empty-conf-dir"); + assertTrue(emptyDir.mkdirs()); + String ddl = sql("t1") + .option(FlinkOptions.PATH, path) + .option(FlinkOptions.HADOOP_CONF_DIR, emptyDir.getAbsolutePath()) + .end(); + + batchTableEnv.executeSql(ddl); + Throwable e = assertThrows(Exception.class, + () -> batchTableEnv.sqlQuery("select * from t1").execute().collect().next()); + assertTrue(getRootMessage(e).contains(emptyDir.getAbsolutePath()), + "Error should name the invalid conf dir, got: " + getRootMessage(e)); + } + + /** + * 'hadoop.*' single-property options take precedence over the site files + * loaded from 'hadoop_conf.dir'. + */ + @Test + public void testHadoopOptionOverridesConfDir() throws Exception { + File hadoopConfDir = createHadoopConfDirWithRemoteFs(); + StoragePath path = new StoragePath(tempFile.toURI()); + writeSiteXml(new File(hadoopConfDir, "core-site.xml"), + "test.prop", "site-value"); + + String ddl = sql("t1") + .option(FlinkOptions.PATH, path) + .option(FlinkOptions.HADOOP_CONF_DIR, hadoopConfDir.getAbsolutePath()) + .option("hadoop.test.prop", "override-value") + .end(); + batchTableEnv.executeSql(ddl); + + // verify via HadoopConfigurations directly since the DDL path only + // exercises the conf inside the table source + Configuration flinkConf = new Configuration(); + flinkConf.setString(FlinkOptions.HADOOP_CONF_DIR.key(), + hadoopConfDir.getAbsolutePath()); + flinkConf.setString("hadoop.test.prop", "override-value"); + + org.apache.hadoop.conf.Configuration hadoopConf = + HadoopConfigurations.getHadoopConf(flinkConf); + assertEquals("override-value", hadoopConf.get("test.prop"), + "hadoop.* options must override values from the conf dir"); + } + @ParameterizedTest @MethodSource("indexAndPartitioningParams") void testWriteMultipleCommitWithDifferentLogBlockType(String indexType, boolean hiveStylePartitioning) throws Exception { @@ -4358,6 +4510,56 @@ private List execSelectSql(TableEnvironment tEnv, String select, long timeo .collect(Collectors.toList()); } + /** Local-backed remote filesystem with eagerly loaded permissions for recursive listings. */ + public static class ConfDirFileSystem extends NonLocalSchemeLocalFileSystem { + @Override + public org.apache.hadoop.fs.FileStatus getFileStatus(org.apache.hadoop.fs.Path path) throws IOException { + org.apache.hadoop.fs.FileStatus status = super.getFileStatus(path); + // RawLocalFileSystem loads permissions lazily using java.io.File, which requires a file URI. + status.setPath(new org.apache.hadoop.fs.Path(new File(path.toUri().getPath()).toURI())); + org.apache.hadoop.fs.FileStatus copy = new org.apache.hadoop.fs.FileStatus(status); + copy.setPath(makeQualified(path)); + return copy; + } + } + + /** Registers a non-local filesystem that cannot be resolved without the supplied configuration. */ + private File createHadoopConfDirWithRemoteFs() throws IOException { + File dir = new File(tempFile, "remote-hadoop-conf"); + assertTrue(dir.mkdirs()); + org.apache.hadoop.conf.Configuration remoteConf = new org.apache.hadoop.conf.Configuration(false); + remoteConf.set("fs.viewfs.impl", ConfDirFileSystem.class.getName()); + remoteConf.setBoolean("fs.viewfs.impl.disable.cache", true); + try (FileWriter writer = new FileWriter(new File(dir, "core-site.xml"))) { + remoteConf.writeXml(writer); + } + writeSiteXml(new File(dir, "hdfs-site.xml"), "dfs.replication", "1"); + return dir; + } + + private static void writeSiteXml(File file, String name, String value) + throws IOException { + String content = "\n" + + "\n" + + " " + name + "" + value + + "\n" + + "\n"; + try (FileWriter w = new FileWriter(file)) { + w.write(content); + } + } + + private static String getRootMessage(Throwable t) { + StringBuilder sb = new StringBuilder(); + while (t != null) { + if (t.getMessage() != null) { + sb.append(t.getMessage()).append("; "); + } + t = t.getCause(); + } + return sb.toString(); + } + private List fetchResultWithExpectedNum(TableEnvironment tEnv, TableResult tableResult) { try { // wait the continuous streaming query to be terminated by forced exception with expected row number