[INLONG-3815][Sort] Fix meta field sync error (#4055)

Author: yunqingmoswu

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/format/CanalJsonFormat.java

@@ -76,7 +76,7 @@ public CanalJsonFormat() {
     @JsonIgnore
     @Override
     public String getFormat() {
-        return "canal-json";
+        return "canal-json-inlong";
     }
 
     /**
@@ -89,13 +89,13 @@ public Map<String, String> generateOptions() {
         Map<String, String> options = new HashMap<>(16);
         options.put("format", getFormat());
         if (this.ignoreParseErrors != null) {
-            options.put("canal-json.ignore-parse-errors", this.ignoreParseErrors.toString());
+            options.put("canal-json-inlong.ignore-parse-errors", this.ignoreParseErrors.toString());
         }
-        options.put("canal-json.timestamp-format.standard", this.timestampFormatStandard);
-        options.put("canal-json.map-null-key.mode", this.mapNullKeyMode);
-        options.put("canal-json.map-null-key.literal", this.mapNullKeyLiteral);
+        options.put("canal-json-inlong.timestamp-format.standard", this.timestampFormatStandard);
+        options.put("canal-json-inlong.map-null-key.mode", this.mapNullKeyMode);
+        options.put("canal-json-inlong.map-null-key.literal", this.mapNullKeyLiteral);
         if (this.encodeDecimalAsPlainNumber != null) {
-            options.put("canal-json.encode.decimal-as-plain-number", this.encodeDecimalAsPlainNumber.toString());
+            options.put("canal-json-inlong.encode.decimal-as-plain-number", this.encodeDecimalAsPlainNumber.toString());
         }
         return options;
     }

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/KafkaLoadNode.java

@@ -110,7 +110,7 @@ public Map<String, String> tableOptions() {
                 options.putAll(format.generateOptions(true));
             }
         } else if (format instanceof CanalJsonFormat || format instanceof DebeziumJsonFormat) {
-            options.put("connector", "kafka");
+            options.put("connector", "kafka-inlong");
             options.putAll(format.generateOptions(false));
         } else {
             throw new IllegalArgumentException("kafka load Node format is IllegalArgument");

inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/parser/impl/FlinkSqlParser.java

@@ -628,34 +628,34 @@ private String parseKafkaLoadNodeMetaField(BuiltInFieldInfo metaField) {
                 metaType = "STRING METADATA FROM 'value.database'";
                 break;
             case MYSQL_METADATA_EVENT_TIME:
-                metaType = "TIMESTAMP(3) METADATA FROM 'value.op_ts'";
+                metaType = "TIMESTAMP(3) METADATA FROM 'value.event-timestamp'";
                 break;
             case MYSQL_METADATA_EVENT_TYPE:
-                metaType = "STRING METADATA FROM 'value.op_type'";
+                metaType = "STRING METADATA FROM 'value.op-type'";
                 break;
             case MYSQL_METADATA_DATA:
                 metaType = "STRING METADATA FROM 'value.data'";
                 break;
             case MYSQL_METADATA_IS_DDL:
-                metaType = "BOOLEAN METADATA FROM 'value.is_ddl'";
+                metaType = "BOOLEAN METADATA FROM 'value.is-ddl'";
                 break;
             case METADATA_TS:
-                metaType = "TIMESTAMP_LTZ(3) METADATA FROM 'value.ts'";
+                metaType = "TIMESTAMP_LTZ(3) METADATA FROM 'value.ingestion-timestamp'";
                 break;
             case METADATA_SQL_TYPE:
-                metaType = "MAP<STRING, INT> METADATA FROM 'value.sql_type'";
+                metaType = "MAP<STRING, INT> METADATA FROM 'value.sql-type'";
                 break;
             case METADATA_MYSQL_TYPE:
-                metaType = "MAP<STRING, STRING> METADATA FROM 'value.mysql_type'";
+                metaType = "MAP<STRING, STRING> METADATA FROM 'value.mysql-type'";
                 break;
             case METADATA_PK_NAMES:
-                metaType = "ARRAY<STRING> METADATA FROM 'value.pk_names'";
+                metaType = "ARRAY<STRING> METADATA FROM 'value.pk-names'";
                 break;
             case METADATA_BATCH_ID:
-                metaType = "BIGINT METADATA FROM 'value.batch_id'";
+                metaType = "BIGINT METADATA FROM 'value.batch-id'";
                 break;
             case METADATA_UPDATE_BEFORE:
-                metaType = "ARRAY<MAP<STRING, STRING>> METADATA FROM 'value.update_before'";
+                metaType = "ARRAY<MAP<STRING, STRING>> METADATA FROM 'value.update-before'";
                 break;
             default:
                 metaType = TableFormatUtils.deriveLogicalType(metaField.getFormatInfo()).asSummaryString();

inlong-sort/sort-single-tenant/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory

@@ -32,3 +32,4 @@
 
 org.apache.inlong.sort.singletenant.flink.cdc.mysql.table.MySqlTableInlongSourceFactory
 org.apache.inlong.sort.singletenant.flink.connectors.hive.table.catalog.factories.HiveTableInlongFactory
+org.apache.inlong.sort.flink.kafka.KafkaDynamicTableFactory