This is an automated email from the ASF dual-hosted git repository.
danny0405 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 ecb7e8ff3893 feat(flink): support per-table Hadoop configuration
directories (#20433)
ecb7e8ff3893 is described below
commit ecb7e8ff389377f92dbc133e097508f9e62623cc
Author: cbg-wx <[email protected]>
AuthorDate: Thu Oct 29 18:01:28 2026 +0800
feat(flink): support per-table Hadoop configuration directories (#20433)
* Add Hadoop configuration directory hadoop_conf.dir support for Flink
cross-cluster Hudi read/write
* fix(flink): validate and propagate explicit Hadoop configuration
* refactor(flink): simplify Hadoop configuration directory validation
---------
Co-authored-by: danny0405 <[email protected]>
---
.../hudi/client/HoodieFlinkTableServiceClient.java | 5 +-
.../java/org/apache/hudi/util/FlinkClientUtil.java | 69 ++++--
.../client/TestHoodieFlinkTableServiceClient.java | 6 +
.../org/apache/hudi/util/TestFlinkClientUtil.java | 22 ++
.../apache/hudi/configuration/FlinkOptions.java | 11 +
.../hudi/configuration/HadoopConfigurations.java | 4 +-
.../configuration/TestHadoopConfigurations.java | 272 +++++++++++++++++++++
.../apache/hudi/table/ITTestHoodieDataSource.java | 202 +++++++++++++++
8 files changed, 560 insertions(+), 31 deletions(-)
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 8ec43d9781d4..4d295ad9b50b 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.HoodieClusteringException;
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.HoodieTable;
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 @@ public class HoodieFlinkTableServiceClient<T> extends
BaseHoodieTableServiceClie
*/
private HoodieBackedTableMetadataWriter initMetadataWriter(Option<String>
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 d4e67507b6ae..baa8fe42de7a 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.HoodieTableConfig;
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.Logger;
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 class FlinkClientUtil {
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 class FlinkClientUtil {
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 7ddacba98a66..7f9ffeacae47 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.assertNull;
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 @@ public class TestHoodieFlinkTableServiceClient extends
HoodieFlinkClientTestHarn
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 cc4df7c103e4..abdd5d6b050d 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 @@ package org.apache.hudi.util;
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 class TestFlinkClientUtil {
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"),
"<configuration><property><name>fs.defaultFS</name>"
+ + "<value>hdfs://remote-cluster</value></property><property><name>" +
CRYPTO_RETRIEVER_KEY
+ + "</name><value>site-value</value></property></configuration>");
+ try (MockedStatic<GlobalConfiguration> 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 6ec2adea7116..e0758e447731 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<String> 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 12d920ac9870..0a10338e8183 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 class HadoopConfigurations {
* 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<String, String> 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 000000000000..078c05cb0b62
--- /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 "<?xml version=\"1.0\"?>\n"
+ + "<configuration>\n"
+ + " <property>\n"
+ + " <name>" + name + "</name>\n"
+ + " <value>" + value + "</value>\n"
+ + " </property>\n"
+ + "</configuration>\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 5740a5e2e6d3..c877fda22097 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.common.util.CollectionUtils;
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.MethodSource;
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 @@ public class ITTestHoodieDataSource {
+ "+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<Row> 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<Row> 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 @@ public class ITTestHoodieDataSource {
.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 = "<?xml version=\"1.0\"?>\n"
+ + "<configuration>\n"
+ + " <property><name>" + name + "</name><value>" + value
+ + "</value></property>\n"
+ + "</configuration>\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<Row> fetchResultWithExpectedNum(TableEnvironment tEnv,
TableResult tableResult) {
try {
// wait the continuous streaming query to be terminated by forced
exception with expected row number
