hudi-agent commented on code in PR #19392:
URL: https://github.com/apache/hudi/pull/19392#discussion_r3670926646
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java:
##########
@@ -135,7 +135,7 @@ public static TypedProperties
appendKafkaProps(FlinkStreamerConfig config) {
public static TypedProperties getProps(FlinkStreamerConfig cfg) {
if (cfg.propsFilePath.isEmpty()) {
- return new TypedProperties();
+ return buildProperties(cfg.configs);
Review Comment:
π€ Nice fix β this makes `--hoodie-conf` actually reach `toFlinkConfig` (line
420) when no `--props` is set. One thing worth confirming: `buildProperties`
also pulls in `DFSPropertiesConfiguration.getGlobalProps()`
(hudi-defaults.conf), so the no-props path now seeds the streamer writer config
with global defaults where it previously returned empty props. The `--props`
branch (`new DFSPropertiesConfiguration(hadoopConf, path)`) doesn't load
globals, so the two branches diverge. Is applying global defaults here intended?
<sub><i>β οΈ AI-generated; verify before applying. React π/π to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java:
##########
@@ -270,4 +283,136 @@ void testEstimateFileGroupCountForGlobalRLI() {
conf.setString(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP.key(),
"11");
assertEquals(11, OptionsResolver.estimateFileGroupCountForRLI(conf));
}
+
+ @Test
+ void testIncrementalJobGraphPredicate() {
+ Configuration conf = new Configuration();
+ assertFalse(OptionsResolver.isIncrementalJobGraph(conf));
+ conf.set(FlinkOptions.WRITE_INCREMENTAL_JOB_GRAPH_GENERATION, true);
+ assertTrue(OptionsResolver.isIncrementalJobGraph(conf));
+ }
+
+ @Test
+ void testTableTypePredicates() {
+ Configuration conf = new Configuration();
+ assertTrue(OptionsResolver.isCowTable(conf));
+ assertFalse(OptionsResolver.isMorTable(conf));
+ assertFalse(OptionsResolver.isMorTable(Collections.emptyMap()));
+ conf.set(FlinkOptions.TABLE_TYPE,
HoodieTableType.MERGE_ON_READ.name().toLowerCase());
+ assertTrue(OptionsResolver.isMorTable(conf));
+ assertTrue(OptionsResolver.isMorTable(
+ Collections.singletonMap(FlinkOptions.TABLE_TYPE.key(),
HoodieTableType.MERGE_ON_READ.name())));
+ }
+
+ @Test
+ void testOperationTypePredicates() {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.OPERATION, WriteOperationType.INSERT.value());
+ assertTrue(OptionsResolver.isInsertOperation(conf));
+ conf.set(FlinkOptions.OPERATION, WriteOperationType.UPSERT.value());
+ assertTrue(OptionsResolver.isUpsertOperation(conf));
+ conf.set(FlinkOptions.OPERATION, WriteOperationType.BULK_INSERT.value());
+ assertTrue(OptionsResolver.isBulkInsertOperation(conf));
+ }
+
+ @Test
+ void testPayloadAndCompactionPredicates() {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.PAYLOAD_CLASS_NAME,
DefaultHoodieRecordPayload.class.getName());
+ assertTrue(OptionsResolver.isDefaultHoodieRecordPayloadClazz(conf));
+ conf.set(FlinkOptions.COMPACTION_TRIGGER_STRATEGY,
FlinkOptions.TIME_ELAPSED.toUpperCase());
+ assertTrue(OptionsResolver.isDeltaTimeCompaction(conf));
+ conf.set(FlinkOptions.COMPACTION_TRIGGER_STRATEGY,
FlinkOptions.NUM_COMMITS);
+ assertFalse(OptionsResolver.isDeltaTimeCompaction(conf));
+ }
+
+ @Test
+ void testReadOptions() {
+ Configuration conf = new Configuration();
+ assertEquals(-1, OptionsResolver.getReadCommitsLimit(conf));
+ conf.set(FlinkOptions.READ_COMMITS_LIMIT, 5);
+ assertEquals(5, OptionsResolver.getReadCommitsLimit(conf));
+
+ conf.set(FlinkOptions.SUPPLEMENTAL_LOGGING_MODE,
+
HoodieCDCSupplementalLoggingMode.DATA_BEFORE_AFTER.name().toLowerCase());
+ assertEquals(HoodieCDCSupplementalLoggingMode.DATA_BEFORE_AFTER,
+ OptionsResolver.getCDCSupplementalLoggingMode(conf));
+
+ conf.set(FlinkOptions.READ_CDC_FROM_CHANGELOG, false);
+ assertFalse(OptionsResolver.readCDCFromChangelog(conf));
+ conf.set(FlinkOptions.RECORD_KEY_FIELD, "id,tenant");
+ assertEquals("id", OptionsResolver.getIndexKeyFields(conf).get(0));
+ assertEquals("tenant", OptionsResolver.getIndexKeyFields(conf).get(1));
Review Comment:
π€ nit: `testWriteOptions` is similarly broad β it covers empty-commit
allowance, two concurrency modes, and key encoding. Something like
`testConcurrencyModeAndWriteFlags` would make the intent clearer without
requiring a full read of the method body.
<sub><i>β οΈ AI-generated; verify before applying. React π/π to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/streamer/TestFlinkStreamerConfig.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.streamer;
+
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.configuration.FlinkOptions;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.ParameterException;
+import org.apache.flink.configuration.Configuration;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link FlinkStreamerConfig}.
+ */
+class TestFlinkStreamerConfig {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void testParseAndDeriveFlinkConfiguration() {
+ FlinkStreamerConfig config = parse(
+ "--kafka-topic", "orders",
+ "--kafka-group-id", "flink-writers",
+ "--kafka-bootstrap-servers", "broker:9092",
+ "--target-base-path", tempDir.toString(),
+ "--target-table", "orders_hudi",
+ "--table-type", "merge_on_read",
+ "--op", "INSERT",
+ "--record-key-field", "order_id",
+ "--partition-path-field", "order_date",
+ "--source-ordering-fields", "event_ts,seq_no",
+ "--instant-retry-times", "7",
+ "--instant-retry-interval", "250",
+ "--filter-dupes",
+ "--commit-on-errors",
+ "--metadata-enabled",
+ "--write-rate-limit", "500",
+ "--write-task-num", "6",
+ "--bucket-assign-num", "5",
+ "--index-bootstrap-num", "4",
+ "--source-avro-schema-path", "file:///tmp/source.avsc",
+ "--source-avro-schema",
"{\"type\":\"record\",\"name\":\"order\",\"fields\":[]}",
+ "--compaction-tasks", "3",
+ "--clustering-tasks", "2",
+ "--hive-sync-enable",
+ "--hive-sync-db", "analytics",
+ "--hive-sync-table", "orders",
+ "--hoodie-conf",
"hoodie.datasource.write.drop.partition.columns=true");
+
+ Configuration conf = FlinkStreamerConfig.toFlinkConfig(config);
+
+ assertEquals(tempDir.toString(), conf.get(FlinkOptions.PATH));
+ assertEquals("orders_hudi", conf.get(FlinkOptions.TABLE_NAME));
+ assertEquals("MERGE_ON_READ", conf.get(FlinkOptions.TABLE_TYPE));
+ assertEquals(WriteOperationType.INSERT.value(),
conf.get(FlinkOptions.OPERATION));
+ assertEquals("order_id", conf.get(FlinkOptions.RECORD_KEY_FIELD));
+ assertEquals("order_date", conf.get(FlinkOptions.PARTITION_PATH_FIELD));
+ assertEquals("event_ts,seq_no", conf.get(FlinkOptions.ORDERING_FIELDS));
+ assertEquals(7, conf.get(FlinkOptions.RETRY_TIMES));
+ assertEquals(250L, conf.get(FlinkOptions.RETRY_INTERVAL_MS));
+ assertTrue(conf.get(FlinkOptions.PRE_COMBINE));
+ assertTrue(conf.get(FlinkOptions.IGNORE_FAILED));
+ assertTrue(conf.get(FlinkOptions.METADATA_ENABLED));
+ assertEquals(500L, conf.get(FlinkOptions.WRITE_RATE_LIMIT));
+ assertEquals(6, conf.get(FlinkOptions.WRITE_TASKS));
+ assertEquals(5, conf.get(FlinkOptions.BUCKET_ASSIGN_TASKS));
+ assertEquals(4, conf.get(FlinkOptions.INDEX_BOOTSTRAP_TASKS));
+ assertEquals("file:///tmp/source.avsc",
conf.get(FlinkOptions.SOURCE_AVRO_SCHEMA_PATH));
+ assertEquals(3, conf.get(FlinkOptions.COMPACTION_TASKS));
+ assertEquals(2, conf.get(FlinkOptions.CLUSTERING_TASKS));
+ assertTrue(conf.get(FlinkOptions.HIVE_SYNC_ENABLED));
+ assertEquals("analytics", conf.get(FlinkOptions.HIVE_SYNC_DB));
+ assertEquals("orders", conf.get(FlinkOptions.HIVE_SYNC_TABLE));
+ assertEquals("hoodie.datasource.write.drop.partition.columns=true",
config.configs.get(0));
Review Comment:
π€ Tracing this path, `config.configs` does seem to reach the derived
`Configuration` even without `--props`: `StreamerUtil.getProps()` returns
`buildProperties(cfg.configs)` when `propsFilePath` is empty, and
`toFlinkConfig()` feeds that map straight into `fromMap()`. So `--hoodie-conf
drop.partition.columns=true` should land in the derived config. I agree
asserting the value here is worth adding for coverage β I'd just expect it to
pass as-is rather than require a fix to that path. Worth double-checking
against what you observed.
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/streamer/TestFlinkStreamerConfig.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.streamer;
+
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.configuration.FlinkOptions;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.ParameterException;
+import org.apache.flink.configuration.Configuration;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link FlinkStreamerConfig}.
+ */
+class TestFlinkStreamerConfig {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void testParseAndDeriveFlinkConfiguration() {
+ FlinkStreamerConfig config = parse(
+ "--kafka-topic", "orders",
+ "--kafka-group-id", "flink-writers",
+ "--kafka-bootstrap-servers", "broker:9092",
+ "--target-base-path", tempDir.toString(),
+ "--target-table", "orders_hudi",
+ "--table-type", "merge_on_read",
+ "--op", "INSERT",
+ "--record-key-field", "order_id",
+ "--partition-path-field", "order_date",
+ "--source-ordering-fields", "event_ts,seq_no",
+ "--instant-retry-times", "7",
+ "--instant-retry-interval", "250",
+ "--filter-dupes",
+ "--commit-on-errors",
+ "--metadata-enabled",
+ "--write-rate-limit", "500",
+ "--write-task-num", "6",
+ "--bucket-assign-num", "5",
+ "--index-bootstrap-num", "4",
+ "--source-avro-schema-path", "file:///tmp/source.avsc",
+ "--source-avro-schema",
"{\"type\":\"record\",\"name\":\"order\",\"fields\":[]}",
+ "--compaction-tasks", "3",
+ "--clustering-tasks", "2",
+ "--hive-sync-enable",
+ "--hive-sync-db", "analytics",
+ "--hive-sync-table", "orders",
+ "--hoodie-conf",
"hoodie.datasource.write.drop.partition.columns=true");
+
+ Configuration conf = FlinkStreamerConfig.toFlinkConfig(config);
+
+ assertEquals(tempDir.toString(), conf.get(FlinkOptions.PATH));
+ assertEquals("orders_hudi", conf.get(FlinkOptions.TABLE_NAME));
+ assertEquals("MERGE_ON_READ", conf.get(FlinkOptions.TABLE_TYPE));
+ assertEquals(WriteOperationType.INSERT.value(),
conf.get(FlinkOptions.OPERATION));
+ assertEquals("order_id", conf.get(FlinkOptions.RECORD_KEY_FIELD));
+ assertEquals("order_date", conf.get(FlinkOptions.PARTITION_PATH_FIELD));
+ assertEquals("event_ts,seq_no", conf.get(FlinkOptions.ORDERING_FIELDS));
+ assertEquals(7, conf.get(FlinkOptions.RETRY_TIMES));
+ assertEquals(250L, conf.get(FlinkOptions.RETRY_INTERVAL_MS));
Review Comment:
π€ The mapping does convert here β `FlinkStreamerConfig#toFlinkConfig` sets
`RETRY_INTERVAL_MS` via
`TimeUnit.SECONDS.toMillis(Long.parseLong(config.instantRetryInterval))`
(around line 437-438), so the seconds-valued CLI option is translated to the
ms-valued internal option. So the secondsβms contract does appear consistent
rather than a real unit bug; asserting the converted ms value in the test would
make that translation explicit though.
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java:
##########
@@ -270,4 +283,136 @@ void testEstimateFileGroupCountForGlobalRLI() {
conf.setString(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_MAX_FILE_GROUP_COUNT_PROP.key(),
"11");
assertEquals(11, OptionsResolver.estimateFileGroupCountForRLI(conf));
}
+
+ @Test
+ void testIncrementalJobGraphPredicate() {
+ Configuration conf = new Configuration();
+ assertFalse(OptionsResolver.isIncrementalJobGraph(conf));
+ conf.set(FlinkOptions.WRITE_INCREMENTAL_JOB_GRAPH_GENERATION, true);
+ assertTrue(OptionsResolver.isIncrementalJobGraph(conf));
+ }
+
+ @Test
+ void testTableTypePredicates() {
+ Configuration conf = new Configuration();
+ assertTrue(OptionsResolver.isCowTable(conf));
+ assertFalse(OptionsResolver.isMorTable(conf));
+ assertFalse(OptionsResolver.isMorTable(Collections.emptyMap()));
+ conf.set(FlinkOptions.TABLE_TYPE,
HoodieTableType.MERGE_ON_READ.name().toLowerCase());
+ assertTrue(OptionsResolver.isMorTable(conf));
+ assertTrue(OptionsResolver.isMorTable(
+ Collections.singletonMap(FlinkOptions.TABLE_TYPE.key(),
HoodieTableType.MERGE_ON_READ.name())));
+ }
+
+ @Test
+ void testOperationTypePredicates() {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.OPERATION, WriteOperationType.INSERT.value());
+ assertTrue(OptionsResolver.isInsertOperation(conf));
+ conf.set(FlinkOptions.OPERATION, WriteOperationType.UPSERT.value());
+ assertTrue(OptionsResolver.isUpsertOperation(conf));
+ conf.set(FlinkOptions.OPERATION, WriteOperationType.BULK_INSERT.value());
+ assertTrue(OptionsResolver.isBulkInsertOperation(conf));
+ }
+
+ @Test
+ void testPayloadAndCompactionPredicates() {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.PAYLOAD_CLASS_NAME,
DefaultHoodieRecordPayload.class.getName());
+ assertTrue(OptionsResolver.isDefaultHoodieRecordPayloadClazz(conf));
+ conf.set(FlinkOptions.COMPACTION_TRIGGER_STRATEGY,
FlinkOptions.TIME_ELAPSED.toUpperCase());
+ assertTrue(OptionsResolver.isDeltaTimeCompaction(conf));
+ conf.set(FlinkOptions.COMPACTION_TRIGGER_STRATEGY,
FlinkOptions.NUM_COMMITS);
Review Comment:
π€ nit: `testReadOptions` covers three fairly different things β commit
limits, CDC supplemental logging mode, and index key field parsing. Could you
split these or at least rename to something like
`testReadCommitsLimitAndCdcOptions` so it's clearer what behaviors live here?
<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]
