feat(flink): support per-table Hadoop configuration directories

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<String> latestPendingInstant) {
     return (HoodieBackedTableMetadataWriter) FlinkHoodieBackedTableMetadataWriter.create(
-        HadoopFSUtils.getStorageConf(FlinkClientUtil.getHadoopConf()), this.config, HoodieFlinkEngineContext.DEFAULT, latestPendingInstant);
+        storageConf, this.config, context, latestPendingInstant);
   }
 
   public void initMetadataTable() {

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;
   }
 
   /**

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();
     }

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"), "<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"));
+    }
+  }
+
 }

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
   // -------------------------------------------------------------------------

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<String, String> hadoopOptions = FlinkOptions.getPropertiesWithPrefix(conf.toMap(), HADOOP_PREFIX);
     hadoopOptions.forEach(hadoopConf::set);
     // kind of hacky: flink specific IO options.