@Override
        public Object read(SourceRecord record, @Nullable TableChanges.TableChange tableSchema, RowData rowData) {
            // construct debezium json
            Struct messageStruct = (Struct) record.value();
            Struct sourceStruct = messageStruct.getStruct(FieldName.SOURCE);
            GenericRowData data = (GenericRowData) rowData;
            Map<String, Object> field = (Map<String, Object>) data.getField(0);

            Source source = Source.builder().db(getMetaData(record, AbstractSourceInfo.DATABASE_NAME_KEY))
                    .table(getMetaData(record, AbstractSourceInfo.TABLE_NAME_KEY))
                    .name(sourceStruct.getString(AbstractSourceInfo.SERVER_NAME_KEY)).sqlType(getSqlType(tableSchema))
                    .pkNames(getPkNames(tableSchema)).build();
            DebeziumJson debeziumJson = DebeziumJson.builder().after(field).source(source)
                    .tsMs(sourceStruct.getInt64(AbstractSourceInfo.TIMESTAMP_KEY)).op(getDebeziumOpType(data))
                    .tableChange(tableSchema).build();

            try {
                return StringData.fromString(OBJECT_MAPPER.writeValueAsString(debeziumJson));
            } catch (Exception e) {
                throw new IllegalStateException("exception occurs when get meta data", e);
            }
        }
It indicates the data streams with debezium-json format.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/table/PostgreSQLReadableMetaData.java