modify Validates the options of the factory

apache · Oct 6, 2023 · 095ca3c · 095ca3c
1 parent d3dbacc
commit 095ca3c
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/...tors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicTableFactory.java b/...tors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicTableFactory.java
@@ -189,10 +189,12 @@ public DynamicTableSource createDynamicTableSource(Context context) {
  final DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat =
  getValueDecodingFormat(helper);
 
- helper.validateExcept(PROPERTIES_PREFIX);
-
  final ReadableConfig tableOptions = helper.getOptions();
 
+ final String valueFormatPrefix = tableOptions.getOptional(FORMAT).orElse(tableOptions.get(VALUE_FORMAT));
+ LOG.info("valueFormatPrefix is {}", valueFormatPrefix);
+ helper.validateExcept(PROPERTIES_PREFIX, Constants.DIRTY_PREFIX, valueFormatPrefix);
+
  validateTableSourceOptions(tableOptions);
 
  validatePKConstraints(

diff --git a/...afka/src/main/java/org/apache/inlong/sort/kafka/table/UpsertKafkaDynamicTableFactory.java b/...afka/src/main/java/org/apache/inlong/sort/kafka/table/UpsertKafkaDynamicTableFactory.java
@@ -124,7 +124,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
  helper.discoverDecodingFormat(DeserializationFormatFactory.class, VALUE_FORMAT);
 
  // Validate the option data type.
- helper.validateExcept(PROPERTIES_PREFIX);
+ helper.validateExcept(PROPERTIES_PREFIX, Constants.DIRTY_PREFIX);
  validateSource(
  tableOptions,
  keyDecodingFormat,