danny0405 commented on code in PR #19519:
URL: https://github.com/apache/hudi/pull/19519#discussion_r3892438836
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java:
##########
@@ -2263,6 +2264,85 @@ void testReadWithParquetPredicatePushDown() {
+ "+I[id8, Han, 56, 1970-01-01T00:00:08, par4]]");
}
+ /**
+ * Test that Flink can write and read a Hudi table when hadoop.conf.dir is
explicitly set.
+ * This simulates cross-cluster access by pointing hadoop.conf.dir to a
local conf directory
+ * containing core-site.xml with fs.defaultFS=file:///.
+ */
+ @Test
+ void testBatchWriteAndReadWithHadoopConfDir() throws IOException {
+ // Prepare a hadoop conf dir with core-site.xml pointing to local
filesystem
+ File hadoopConfDir = new File(tempFile.getParentFile(), "hadoop-conf");
+ hadoopConfDir.mkdirs();
+ String coreSiteXml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+ + "<configuration>\n"
+ + "
<property><name>fs.defaultFS</name><value>file:///</value></property>\n"
+ + "</configuration>";
+ try (FileWriter w = new FileWriter(new File(hadoopConfDir,
"core-site.xml"))) {
+ w.write(coreSiteXml);
+ }
+
+ TableEnvironment tableEnv = batchTableEnv;
+ String hoodieTableDDL = sql("t1")
+ .option(FlinkOptions.PATH, tempFile.toURI().toString())
+ // specify hadoop.conf.dir to simulate cross-cluster configuration
+ .option(FlinkOptions.HADOOP_CONF_DIR,
hadoopConfDir.getAbsolutePath().replace('\\', '/'))
+ .option(FlinkOptions.METADATA_ENABLED, false)
+ .end();
+ tableEnv.executeSql(hoodieTableDDL);
+
+ execInsertSql(tableEnv, TestSQL.INSERT_T1);
+
+ List<Row> result = CollectionUtil.iterableToList(
+ () -> tableEnv.sqlQuery("select * from t1").execute().collect());
+ assertRowsEquals(result, TestData.DATA_SET_SOURCE_INSERT);
+ }
+
+ /**
+ * Test incremental read with hadoop.conf.dir set.
+ * Verifies that the hadoop.conf.dir option is correctly propagated to the
+ * incremental read path (HoodieTableSource ->
HadoopConfigurations.getHadoopConf).
+ */
+ @ParameterizedTest
+ @EnumSource(value = HoodieTableType.class)
+ void testIncrementalReadWithHadoopConfDir(HoodieTableType tableType) throws
Exception {
+ // Prepare a hadoop conf dir with core-site.xml
+ File hadoopConfDir = new File(tempFile.getParentFile(),
"hadoop-conf-incr");
+ hadoopConfDir.mkdirs();
+ String coreSiteXml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+ + "<configuration>\n"
+ + "
<property><name>fs.defaultFS</name><value>file:///</value></property>\n"
+ + "</configuration>";
+ try (FileWriter w = new FileWriter(new File(hadoopConfDir,
"core-site.xml"))) {
+ w.write(coreSiteXml);
+ }
+ String tablePath = tempFile.toURI().toString();
+ // Step 1: write first batch
+ Configuration conf = TestConfigurations.getDefaultConf(tablePath);
+ conf.set(FlinkOptions.TABLE_TYPE, tableType.name());
+ TestData.writeData(TestData.DATA_SET_INSERT, conf);
+
+ String firstCommit = TestUtils.getFirstCompleteInstant(tablePath);
+
+ // Step 2: write second batch
+ TestData.writeData(TestData.DATA_SET_UPDATE_INSERT, conf);
+
+ // Step 3: incremental read from firstCommit with hadoop.conf.dir set
+ TableEnvironment tableEnv = batchTableEnv;
+ String hoodieTableDDL = sql("t1")
+ .option(FlinkOptions.PATH, tablePath)
+ .option(FlinkOptions.TABLE_TYPE, tableType)
+ .option(FlinkOptions.QUERY_TYPE,
FlinkOptions.QUERY_TYPE_INCREMENTAL)
+ .option(FlinkOptions.READ_START_COMMIT, firstCommit)
+ .option(FlinkOptions.HADOOP_CONF_DIR,
hadoopConfDir.getAbsolutePath().replace('\\', '/'))
+ .option(FlinkOptions.METADATA_ENABLED, false)
+ .end();
+ tableEnv.executeSql(hoodieTableDDL);
+
+ List<Row> result = CollectionUtil.iterableToList(
+ () -> tableEnv.sqlQuery("select * from t1").execute().collect());
+ // incremental read should return the second batch (update/insert records)
+ assertFalse(result.isEmpty(), "Incremental read should return records
after firstCommit");
Review Comment:
[P1] Close the new test method before the next annotation.
`testIncrementalReadWithHadoopConfDir` is missing its closing `}` after this
assertion, so the following `testWriteWithTimelineServerBasedMarker`
declaration is inside another method. This prevents the test class from
compiling. Please add the closing brace before `@ParameterizedTest`.
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java:
##########
@@ -2263,6 +2264,85 @@ void testReadWithParquetPredicatePushDown() {
+ "+I[id8, Han, 56, 1970-01-01T00:00:08, par4]]");
}
+ /**
+ * Test that Flink can write and read a Hudi table when hadoop.conf.dir is
explicitly set.
+ * This simulates cross-cluster access by pointing hadoop.conf.dir to a
local conf directory
+ * containing core-site.xml with fs.defaultFS=file:///.
+ */
+ @Test
+ void testBatchWriteAndReadWithHadoopConfDir() throws IOException {
+ // Prepare a hadoop conf dir with core-site.xml pointing to local
filesystem
+ File hadoopConfDir = new File(tempFile.getParentFile(), "hadoop-conf");
+ hadoopConfDir.mkdirs();
+ String coreSiteXml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+ + "<configuration>\n"
+ + "
<property><name>fs.defaultFS</name><value>file:///</value></property>\n"
+ + "</configuration>";
+ try (FileWriter w = new FileWriter(new File(hadoopConfDir,
"core-site.xml"))) {
+ w.write(coreSiteXml);
+ }
+
+ TableEnvironment tableEnv = batchTableEnv;
+ String hoodieTableDDL = sql("t1")
+ .option(FlinkOptions.PATH, tempFile.toURI().toString())
+ // specify hadoop.conf.dir to simulate cross-cluster configuration
+ .option(FlinkOptions.HADOOP_CONF_DIR,
hadoopConfDir.getAbsolutePath().replace('\\', '/'))
Review Comment:
[P2] Make the integration test depend on loading the supplied Hadoop
configuration.
The table path explicitly uses `file:`, and the XML only sets
`fs.defaultFS=file:///`. The read/write can therefore succeed even if
`hadoop.conf.dir` is completely ignored. The incremental-read test has the same
limitation. Could we use a filesystem setup that requires a property supplied
only by this directory (for example, a test filesystem implementation
registered through the XML), and verify that removing the directory option
makes the operation fail? That would demonstrate propagation to the actual
read/write path.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java:
##########
@@ -49,7 +49,14 @@ public static org.apache.hadoop.conf.Configuration
getParquetConf(
*/
public static org.apache.hadoop.conf.Configuration
getHadoopConf(Configuration conf) {
org.apache.hadoop.conf.Configuration hadoopConf =
FlinkClientUtil.getHadoopConf();
+ // If hadoop.conf.dir is explicitly configured, load from that directory.
+ // This enables cross-cluster writes by pointing to a different cluster's
Hadoop conf.
+ 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");
Review Comment:
Could we avoid the hardcoded, already-stripped `"conf.dir"` key? Since this
is a new option, my preference is to place it outside the reserved `hadoop.*`
passthrough namespace (for example, `hadoop-conf-dir`), which eliminates the
special case.
If we keep `hadoop.conf.dir`, a small alternative is to exclude the declared
option key before stripping prefixes:
```java
Map<String, String> passthroughOptions = new HashMap<>(conf.toMap());
// Consumed by the connector, not forwarded as a Hadoop property.
passthroughOptions.remove(FlinkOptions.HADOOP_CONF_DIR.key());
FlinkOptions.getPropertiesWithPrefix(passthroughOptions, HADOOP_PREFIX)
.forEach(hadoopConf::set);
```
Also, we can choose between the no-argument and explicit-directory loaders
before constructing `hadoopConf`, instead of loading the environment
configuration first and discarding it when the option is set.
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/FlinkClientUtil.java:
##########
@@ -59,6 +59,15 @@ public static org.apache.hadoop.conf.Configuration
getHadoopConf() {
return hadoopConf;
}
+ public static org.apache.hadoop.conf.Configuration getHadoopConf(String
hadoopConfDir) {
+ org.apache.hadoop.conf.Configuration hadoopConf;
+ hadoopConf = getHadoopConfiguration(hadoopConfDir);
+ if (hadoopConf == null){
Review Comment:
I would make an invalid explicitly configured directory a configuration
error, rather than only logging a warning. A missing/mistyped directory
currently produces `new Configuration()`, so a path that relies on
`fs.defaultFS` can resolve against the default filesystem instead of the
requested cluster, while other paths may lose required cluster settings. Please
validate the explicit directory and fail with its path in the message; reserve
environment discovery for when the option is absent. The new
`testGetHadoopConfWithNonExistentConfDir` should assert that failure instead of
only checking for a non-null result.
--
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]
