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-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
