danny0405 commented on code in PR #19392:
URL: https://github.com/apache/hudi/pull/19392#discussion_r3671119764
##########
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:
Addressed in 5abb17b0d55b. Applying global defaults here was not intended.
The no-props branch now starts from an empty TypedProperties instance and
applies only the CLI hoodie-conf overrides, so it no longer diverges from the
props-file branch by implicitly loading hudi-defaults.conf.
##########
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:
Addressed in 5abb17b0d55b. I split the method into focused tests for the
commit limit, CDC options, and index-key parsing.
--
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]
