Add Hadoop configuration directory support for Flink cross-cluster Hudi read/write

[cbg-wx]

Add hadoop.conf.dir in FlinkOptions to support for Flink cross-cluster Hudi read/write

Describe the issue this Pull Request addresses

This pull request in order to close #19191

Summary and Changelog

1. Add HADOOP_CONF_DIR to FlinkOptions

     @AdvancedConfig
     public static final ConfigOption<String> HADOOP_CONF_DIR = ConfigOptions
             .key("hadoop.conf.dir")
             .stringType()
             .noDefaultValue()
             .withDescription("The hadoop configuration directory, where the core-site.xml, "
                     + "hdfs-site.xml etc. lie in. Used to connect to a different HDFS cluster "
                     + "for cross-cluster writes.");

2. FlinkClientUtil.java — Add an overloaded getHadoopConf(String hadoopConfDir) method with hadoop.conf.dir support.

3. HadoopConfigurations.java — Update getHadoopConf to support hadoop.conf.dir

     public static org.apache.hadoop.conf.Configuration getHadoopConf(Configuration conf) {
       org.apache.hadoop.conf.Configuration hadoopConf = FlinkClientUtil.getHadoopConf();
       String hadoopConfDir = conf.getString(FlinkOptions.HADOOP_CONF_DIR.key(), null);
       if (hadoopConfDir != null && !hadoopConfDir.isEmpty()) {
         hadoopConf = FlinkClientUtil.getHadoopConf(hadoopConfDir);
       }
       Map<String, String> options = FlinkOptions.getPropertiesWithPrefix(conf.toMap(), HADOOP_PREFIX);
       options.remove("conf.dir");
       options.forEach(hadoopConf::set);
       return hadoopConf;
     }

Impact

flink read/write hudi table

Risk Level

Documentation Update

Contributor's checklist

• Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
• Enough context is provided in the sections above
• Adequate tests were added if applicable

[hudi-bot]

CI report:

• c885576 Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[hudi-agent]

⚠️ 🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.

Thanks for working on this! The PR adds a hadoop.conf.dir Flink option so the Hadoop configuration can be loaded from a specified directory for cross-cluster reads/writes. A couple of edge cases worth double-checking in the inline comments — notably the silent fallback when the configured directory doesn't exist, and the config key living inside the hadoop. passthrough prefix. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small readability suggestions below, mainly around the unexplained magic string removal and minor code duplication in the test helpers.

[hudi-agent]

🤖 When the directory doesn't exist (or is mistyped), getHadoopConfiguration returns null and this falls back to a bare new Configuration(). Since HadoopConfigurations.getHadoopConf then replaces the env-derived conf with this result, a typo in hadoop.conf.dir would silently use the default/local cluster config instead of the intended remote one — for a cross-cluster write that could route data to the wrong cluster with no error. Could you log a warning (or fail) when the configured dir doesn't exist rather than silently falling back?

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[hudi-agent]

🤖 The hadoop.conf.dir key falls inside the hadoop. passthrough prefix, so it also gets picked up by getPropertiesWithPrefix(..., HADOOP_PREFIX) as conf.dir, forcing this options.remove("conf.dir") workaround. That coupling is easy to break later. Was naming it outside the prefix (e.g. under a hoodie./flink-specific namespace) considered so no special-casing is needed?

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[hudi-agent]

🤖 nit: the magic string "conf.dir" here is easy to misread — could you add an inline comment explaining that this strips the key left over after the hadoop. prefix is removed from hadoop.conf.dir? Something like // remove "conf.dir" — the residual key after stripping the "hadoop." prefix from hadoop.conf.dir.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[hudi-agent]

🤖 nit: could you combine the declaration and assignment into a single line? org.apache.hadoop.conf.Configuration hadoopConf = getHadoopConfiguration(hadoopConfDir); reads more naturally.

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}

[hudi-agent]

🤖 nit: writeCoreSite and writeHdfsSite are identical except for the filename — have you considered collapsing them into a single writeSiteXml(File dir, String fileName, String key, String value) helper?

_{⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.}