Repository navigation
feat(flink): support per-table Hadoop configuration directories - #20433
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #20433 +/- ##
============================================
+ Coverage 80.82% 80.84% +0.01%
- Complexity 35912 36052 +140
============================================
Files 2575 2575
Lines 146762 147254 +492
Branches 17984 18303 +319
============================================
+ Hits 118627 119046 +419
- Misses 19997 20014 +17
- Partials 8138 8194 +56
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
danny0405
left a comment
danny0405
left a comment
There was a problem hiding this comment.
Reviewed commit 5f702c6 for CI failures, correctness, performance, and simplification. The configuration issues below should be addressed before merging.
CI findings:
- test-flink-1, test-flink-2, integration-tests, and Flink 2.2 bundle validation all stop at Checkstyle in FlinkClientUtil.java:92:
if (hadoopConf == null){. These are build failures before the affected test suites execute, not test assertion failures. Example: https://github.com/apache/hudi/actions/runs/40188806680/job/121155173710 - validate-pr-and-source reports an empty Risk Level and a title missing an accepted prefix. Documentation Update also needs completing. Please provide an accurate risk assessment/documentation plan and use an accepted title convention, such as
feat: .... Log: https://github.com/apache/hudi/actions/runs/40188806688/job/121155173602 - Duplicate imports, a wildcard import, indentation inconsistencies, and
}elsealso need cleaning up. Run Checkstyle across both changed modules after fixing the first error.
Performance: no clear throughput regression found by source inspection. The unset-option path is essentially unchanged. The configured path adds directory validation/listing on each configuration construction; checking the four supported filenames directly would avoid that scan. No benchmarks were run.
Simplification:
- Remove the null fallback in HadoopConfigurations: the new getHadoopConf(String) overload always returns a non-null configuration.
- Correct the comment claiming invalid directories fall back; validation throws instead.
- Share supported filenames and common configuration-overlay logic between validation/loading paths.
Validation scope: CI logs and source inspection only; no local tests or benchmarks were run.
| org.apache.hadoop.conf.Configuration hadoopConf; | ||
| if (hadoopConfDir != null) { | ||
| validateHadoopConfDir(hadoopConfDir); | ||
| org.apache.hadoop.conf.Configuration dirHadoopConf= FlinkClientUtil.getHadoopConf(hadoopConfDir); |
There was a problem hiding this comment.
[P1] Propagate this configuration through metadata initialization too. The main write client receives this configuration, but StreamWriteOperatorCoordinator.start() also invokes initMetadataTable(). HoodieFlinkTableServiceClient.initMetadataWriter() still calls the no-argument FlinkClientUtil.getHadoopConf() and uses HoodieFlinkEngineContext.DEFAULT (HoodieFlinkTableServiceClient.java:188-190). With metadata enabled and a remote HA nameservice defined only in hadoop_conf.dir, the metadata path lacks that configuration. This existing bypass leaves the new feature incomplete. Reuse the client's storage configuration/context and add a metadata-enabled cross-cluster propagation test.
|
|
||
| public static org.apache.hadoop.conf.Configuration getHadoopConf(String hadoopConfDir) { | ||
| org.apache.hadoop.conf.Configuration hadoopConf; | ||
| hadoopConf = getHadoopConfiguration(hadoopConfDir); |
There was a problem hiding this comment.
[P2] Preserve the common Flink Hadoop-property overlay. The existing no-argument method applies prefixed Hadoop properties from FLINK_CONF_DIR at lines 73-83, but this overload bypasses that step. Enabling hadoop_conf.dir therefore drops properties supplied through Flink configuration even when the specified directory does not replace them, potentially losing authentication or filesystem settings. Share the overlay logic, define precedence, and test a property supplied only through Flink configuration.
| "Invalid '%s': the specified path is not a directory: %s", | ||
| optionKey, dir.getAbsolutePath())); | ||
| } | ||
| File[] siteFiles = dir.listFiles((d, name) -> name.endsWith("-site.xml")); |
There was a problem hiding this comment.
[P2] Validate the files the loader actually supports. This accepts any name ending in -site.xml, but getHadoopConfiguration() reads only core-site.xml, hdfs-site.xml, yarn-site.xml, and mapred-site.xml. A directory containing only hive-site.xml passes validation while contributing no configuration, allowing the default/environment filesystem to be used despite the promised fail-fast behavior. Check supported readable regular files, share the filename set with the loader, and test a directory containing only an unsupported site file.
| public void testWriteAndReadWithHadoopConfDir() throws Exception { | ||
| File hadoopConfDir = createHadoopConfDirWithFileFs(); | ||
|
|
||
| StoragePath path = new StoragePath(tempFile.toURI()); |
There was a problem hiding this comment.
[P2] Make the integration tests depend on the supplied configuration. This uses a fully qualified file: table path, and the helper sets fs.defaultFS=file:///. Both positive tests can succeed even when downstream components ignore hadoop_conf.dir. Exercise a filesystem/nameservice usable only with the supplied configuration, including metadata-enabled writes, and assert expected records. Also, testHadoopOptionOverridesConfDir puts hadoop.test.prop only in Flink options, not XML, so it does not test overriding a conflicting XML value.
danny0405
changed the title
danny0405
changed the title
Describe the issue this Pull Request addresses
Closes #19191. Flink tables need to read and write a Hudi table on a different Hadoop cluster without replacing the process-wide Hadoop configuration. The table-specific configuration must also reach metadata initialization.
Summary and Changelog
hadoop_conf.dirtable option for an explicit local Hadoop configuration directory.hadoop.*options last.Impact
Adds a Flink configuration option. When unset, existing environment discovery remains unchanged. The configured directory must exist at the same local path on the client, JobManager and TaskManagers. No storage-format change. Configuration loading checks four known filenames instead of listing the directory; no throughput benchmark was performed.
Risk Level
medium: configuration precedence and metadata initialization affect filesystem access. Focused regression tests cover the inherited Flink properties, invalid directories and metadata configuration/context propagation. Integration validation uses a local-backed non-local filesystem registered only in the supplied directory; this does not replace testing against a real remote Hadoop cluster.
Documentation Update
The Flink option description documents supported files, deployment requirements, fail-fast behavior, and precedence. The corresponding Hudi website configuration documentation still needs a follow-up update in the website repository; it is not included in this code PR.
Contributor's checklist
Testing