[GitHub] [inlong] EMsnap commented on a diff in pull request #6210: [INLONG-6174][Sort] MySql connector support meta data with debezium format

[GitBox <gi...@apache.org>]

EMsnap commented on code in PR #6210:
URL: https://github.com/apache/inlong/pull/6210#discussion_r1001439770


##########
inlong-sort/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/MySqlReadableMetadata.java:
##########
@@ -379,6 +415,38 @@ public Object read(SourceRecord record) {
                 }
             });
 
+    private static StringData getCanalData(SourceRecord record, GenericRowData rowData,
+        TableChange tableSchema) {
+        Struct messageStruct = (Struct) record.value();
+        Struct sourceStruct = messageStruct.getStruct(FieldName.SOURCE);
+        // tableName
+        String tableName = getMetaData(record, AbstractSourceInfo.TABLE_NAME_KEY);
+        // databaseName
+        String databaseName = getMetaData(record, AbstractSourceInfo.DATABASE_NAME_KEY);
+        // opTs
+        long opTs = (Long) sourceStruct.get(AbstractSourceInfo.TIMESTAMP_KEY);
+        // ts
+        long ts = (Long) messageStruct.get(FieldName.TIMESTAMP);
+        // actual data
+        GenericRowData data = rowData;
+        Map<String, Object> field = (Map<String, Object>) data.getField(0);
+        List<Map<String, Object>> dataList = new ArrayList<>();
+        dataList.add(field);
+
+        CanalJson canalJson = CanalJson.builder()
+            .data(dataList).database(databaseName)
+            .sql("").es(opTs).isDdl(false).pkNames(getPkNames(tableSchema))
+            .mysqlType(getMysqlType(tableSchema)).table(tableName).ts(ts)
+            .type(getOpType(record)).sqlType(getSqlType(tableSchema)).build();
+
+        try {
+            ObjectMapper objectMapper = new ObjectMapper();

Review Comment:
   done thanks



-- 
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: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org