hudi-agent commented on code in PR #20285:
URL: https://github.com/apache/hudi/pull/20285#discussion_r4241612764
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java:
##########
@@ -1037,6 +1037,14 @@ private FlinkOptions() {
.defaultValue(HoodieSyncTableStrategy.ALL.name())
.withDescription("Hive table synchronization strategy. Available option:
RO, RT, ALL.");
+ @AdvancedConfig
+ public static final ConfigOption<String> HADOOP_CONF_DIR = ConfigOptions
Review Comment:
🤖 Some paths still build their Hadoop conf from
`FlinkClientUtil.getHadoopConf()` and never see this option:
`HoodieFlinkTableServiceClient.initMetadataWriter`,
`HoodieFlinkEngineContext.DEFAULT` (used by `FlinkTables.createTable(conf)`),
and `ColumnStatsIndices.readColumnStatsIndexByColumns`. So for a remote table,
MDT init and column-stats data skipping would resolve the path against the
local cluster (an unknown nameservice, or the wrong `fs.defaultFS`). Could you
route these through `HadoopConfigurations.getHadoopConf(conf)` as well, or note
the limitation?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java:
##########
@@ -48,12 +51,54 @@ public static org.apache.hadoop.conf.Configuration
getParquetConf(
* Creates a new hadoop configuration that is initialized with the given
flink configuration.
*/
public static org.apache.hadoop.conf.Configuration
getHadoopConf(Configuration conf) {
- org.apache.hadoop.conf.Configuration hadoopConf =
FlinkClientUtil.getHadoopConf();
+ // Choose the base configuration source: explicit conf dir wins;
+ // fall back to environment-based discovery when the option is absent
+ // or the directory is invalid.
+ String hadoopConfDir = conf.getString(FlinkOptions.HADOOP_CONF_DIR.key(),
null);
+ org.apache.hadoop.conf.Configuration hadoopConf;
+ if (hadoopConfDir != null) {
+ validateHadoopConfDir(hadoopConfDir);
Review Comment:
🤖 `getHadoopConf(conf)` also runs inside TaskManager operators and readers
(`BucketAssignFunction`, `BootstrapOperator`, `MergeOnReadInputFormat`,
`CkpMetadata`, ...), so this local `File` check and the site-file load need the
directory on every TM host, not just the client. Is that the intended
deployment model? If so, could you mention it in the option description, since
otherwise jobs would pass submission and then fail at operator open?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java:
##########
@@ -48,12 +51,54 @@ public static org.apache.hadoop.conf.Configuration
getParquetConf(
* Creates a new hadoop configuration that is initialized with the given
flink configuration.
*/
public static org.apache.hadoop.conf.Configuration
getHadoopConf(Configuration conf) {
- org.apache.hadoop.conf.Configuration hadoopConf =
FlinkClientUtil.getHadoopConf();
+ // Choose the base configuration source: explicit conf dir wins;
+ // fall back to environment-based discovery when the option is absent
+ // or the directory is invalid.
+ String hadoopConfDir = conf.getString(FlinkOptions.HADOOP_CONF_DIR.key(),
null);
+ org.apache.hadoop.conf.Configuration hadoopConf;
+ if (hadoopConfDir != null) {
+ validateHadoopConfDir(hadoopConfDir);
+ org.apache.hadoop.conf.Configuration dirHadoopConf=
FlinkClientUtil.getHadoopConf(hadoopConfDir);
+ hadoopConf = dirHadoopConf != null ? dirHadoopConf :
FlinkClientUtil.getHadoopConf();
+ }else {
+ hadoopConf = FlinkClientUtil.getHadoopConf();
+ }
Map<String, String> options =
FlinkOptions.getPropertiesWithPrefix(conf.toMap(), HADOOP_PREFIX);
options.forEach(hadoopConf::set);
return hadoopConf;
}
+ /**
+ * Validates an explicitly configured {@code hadoop_conf.dir}. Any invalid
value must
+ * fail fast rather than silently falling back, because an empty
configuration would
+ * resolve {@code fs.defaultFS} to the local FS and mask mis-configuration.
+ */
+ private static void validateHadoopConfDir(String hadoopConfDir) {
+ String optionKey = FlinkOptions.HADOOP_CONF_DIR.key();
+ if (StringUtils.isNullOrEmpty(hadoopConfDir.trim())) {
+ throw new HoodieValidationException(String.format(
+ "'%s' must not be empty. Either set a valid directory or leave
it unset to use environment discovery",
+ optionKey));
+ }
+ File dir = new File(hadoopConfDir);
+ if (!dir.exists()) {
+ throw new HoodieValidationException(String.format(
+ "Invalid '%s': the specified Hadoop conf directory does not
exist: %s",
+ optionKey, hadoopConfDir));
+ }
+ if (!dir.isDirectory()) {
+ throw new HoodieValidationException(String.format(
+ "Invalid '%s': the specified path is not a directory: %s",
+ optionKey, dir.getAbsolutePath()));
+ }
+ File[] siteFiles = dir.listFiles((d, name) -> name.endsWith("-site.xml"));
Review Comment:
🤖 non-blocking: This check accepts any `*-site.xml` (e.g. only
`hive-site.xml`), but `FlinkClientUtil.getHadoopConfiguration` only loads
core/hdfs/yarn/mapred-site, so a misconfigured dir still passes and falls back
to the classpath `fs.defaultFS`; would it make sense to require at least
`core-site.xml` or `hdfs-site.xml` here?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java:
##########
@@ -48,12 +51,54 @@ public static org.apache.hadoop.conf.Configuration
getParquetConf(
* Creates a new hadoop configuration that is initialized with the given
flink configuration.
*/
public static org.apache.hadoop.conf.Configuration
getHadoopConf(Configuration conf) {
- org.apache.hadoop.conf.Configuration hadoopConf =
FlinkClientUtil.getHadoopConf();
+ // Choose the base configuration source: explicit conf dir wins;
+ // fall back to environment-based discovery when the option is absent
+ // or the directory is invalid.
+ String hadoopConfDir = conf.getString(FlinkOptions.HADOOP_CONF_DIR.key(),
null);
+ org.apache.hadoop.conf.Configuration hadoopConf;
+ if (hadoopConfDir != null) {
+ validateHadoopConfDir(hadoopConfDir);
+ org.apache.hadoop.conf.Configuration dirHadoopConf=
FlinkClientUtil.getHadoopConf(hadoopConfDir);
Review Comment:
🤖 nit: could you drop the `dirHadoopConf != null` fallback?
`FlinkClientUtil.getHadoopConf(String)` never returns null, so the branch is
dead. Also `}else {` and `dirHadoopConf= ` need spaces.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java:
##########
@@ -87,8 +90,7 @@
import static org.apache.hudi.utils.TestData.assertRowsEquals;
import static org.apache.hudi.utils.TestData.map;
import static org.apache.hudi.utils.TestData.row;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.*;
Review Comment:
🤖 nit: could you keep the explicit static imports (`assertEquals`,
`assertFalse`, ...) instead of `Assertions.*`? The rest of the file uses
explicit imports.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
