[Flink] debezium-bson format using the id field in the Kafka Key as Update before information

Author: lizc9

docs/content/cdc-ingestion/debezium-bson.md

@@ -47,7 +47,7 @@ bson-*.jar
 {{< hint info >}}
 The debezium bson format requires insert/update/delete event messages include the full document, and include a field that represents the state of the document before the change.
 This requires setting debezium's capture.mode to change_streams_update_full_with_pre_image and [capture.mode.full.update.type](https://debezium.io/documentation/reference/stable/connectors/mongodb.html#mongodb-property-capture-mode-full-update-type) to post_image.
-The database must be running **MongoDB 6.0 or later** to use this option.
+Before version 6.0 of MongoDB, it was not possible to obtain 'Update Before' information. Therefore, using the id field in the Kafka Key as 'Update before' information
 {{< /hint >}}
 
 Here is a simple example for an update operation captured from a Mongodb customers collection in JSON format:
@@ -145,7 +145,7 @@ Below is a list of top-level field BsonValue conversion examples:
         <td>
             <ul>
                 <li>1735934393769</li>
-                <li>{"$numberLong": 1735934393769}</li>
+                <li>{"$numberLong": "1735934393769"}</li>
             </ul>
         </td>
         <td>"1735934393769"</td>
@@ -186,7 +186,7 @@ Below is a list of top-level field BsonValue conversion examples:
     </tr>
     <tr>
         <td><h5>BsonArray</h5></td>
-        <td>[1,2,{"$numberLong": 1735934393769}]</td>
+        <td>[1,2,{"$numberLong": "1735934393769"}]</td>
         <td>"[1,2,1735934393769]"</td>
     </tr>
     <tr>

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumBsonRecordParser.java

@@ -31,6 +31,8 @@
 
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.TextNode;
 
 import org.bson.BsonDocument;
 import org.bson.BsonValue;
@@ -46,13 +48,16 @@
 import java.util.Map;
 import java.util.Objects;
 
+import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.FIELD_BEFORE;
 import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.FIELD_PAYLOAD;
 import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.FIELD_SCHEMA;
 import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.FIELD_TYPE;
 import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.OP_DELETE;
 import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.OP_INSERT;
 import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.OP_READE;
 import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.OP_UPDATE;
+import static org.apache.paimon.utils.JsonSerdeUtil.fromJson;
+import static org.apache.paimon.utils.JsonSerdeUtil.isNull;
 import static org.apache.paimon.utils.JsonSerdeUtil.writeValueAsString;
 
 /**
@@ -71,8 +76,11 @@ public class DebeziumBsonRecordParser extends DebeziumJsonRecordParser {
 
     private static final String FIELD_COLLECTION = "collection";
     private static final String FIELD_OBJECT_ID = "_id";
+    private static final String FIELD_KEY_ID = "id";
     private static final List<String> PRIMARY_KEYS = Collections.singletonList(FIELD_OBJECT_ID);
 
+    private ObjectNode keyRoot;
+
     public DebeziumBsonRecordParser(TypeMapping typeMapping, List<ComputedColumn> computedColumns) {
         super(typeMapping, computedColumns);
     }
@@ -87,11 +95,11 @@ public List<RichCdcMultiplexRecord> extractRecords() {
                 processRecord(getData(), RowKind.INSERT, records);
                 break;
             case OP_UPDATE:
-                processRecord(getBefore(operation), RowKind.DELETE, records);
+                processDeleteRecord(operation, records);
                 processRecord(getData(), RowKind.INSERT, records);
                 break;
             case OP_DELETE:
-                processRecord(getBefore(operation), RowKind.DELETE, records);
+                processDeleteRecord(operation, records);
                 break;
             default:
                 throw new UnsupportedOperationException("Unknown record operation: " + operation);
@@ -101,11 +109,14 @@ public List<RichCdcMultiplexRecord> extractRecords() {
 
     @Override
     protected void setRoot(CdcSourceRecord record) {
-        JsonNode node = (JsonNode) record.getValue();
-        if (node.has(FIELD_SCHEMA)) {
-            root = node.get(FIELD_PAYLOAD);
-        } else {
-            root = node;
+        root = (JsonNode) record.getValue();
+        if (root.has(FIELD_SCHEMA)) {
+            root = root.get(FIELD_PAYLOAD);
+        }
+
+        keyRoot = (ObjectNode) record.getKey();
+        if (!isNull(keyRoot) && keyRoot.has(FIELD_SCHEMA)) {
+            keyRoot = (ObjectNode) keyRoot.get(FIELD_PAYLOAD);
         }
     }
 
@@ -158,4 +169,37 @@ protected String getTableName() {
     protected String format() {
         return "debezium-bson";
     }
+
+    public boolean checkBeforeExists() {
+        return !isNull(root.get(FIELD_BEFORE));
+    }
+
+    private void processDeleteRecord(String operation, List<RichCdcMultiplexRecord> records) {
+        if (checkBeforeExists()) {
+            processRecord(getBefore(operation), RowKind.DELETE, records);
+        } else {
+            // Before version 6.0 of MongoDB, it was not possible to obtain 'Update Before'
+            // information. Therefore, data is first deleted using the key 'id'
+            JsonNode idNode = null;
+            Preconditions.checkArgument(
+                    !isNull(keyRoot) && !isNull(idNode = keyRoot.get(FIELD_KEY_ID)),
+                    "Invalid %s format: missing '%s' field in key when '%s' is '%s' for: %s.",
+                    format(),
+                    FIELD_KEY_ID,
+                    FIELD_TYPE,
+                    operation,
+                    keyRoot);
+
+            // Deserialize id from json string to JsonNode
+            Map<String, JsonNode> record =
+                    Collections.singletonMap(
+                            FIELD_OBJECT_ID, fromJson(idNode.asText(), JsonNode.class));
+
+            try {
+                processRecord(new TextNode(writeValueAsString(record)), RowKind.DELETE, records);
+            } catch (JsonProcessingException e) {
+                throw new RuntimeException("Failed to deserialize key record.", e);
+            }
+        }
+    }
 }

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumJsonDeserializationSchema.java

@@ -64,7 +64,14 @@ public CdcSourceRecord deserialize(ConsumerRecord<byte[], byte[]> message) throw
         }
 
         try {
-            return new CdcSourceRecord(objectMapper.readValue(message.value(), JsonNode.class));
+            byte[] key = message.key();
+            JsonNode keyNode = null;
+            if (key != null) {
+                keyNode = objectMapper.readValue(key, JsonNode.class);
+            }
+
+            JsonNode valueNode = objectMapper.readValue(message.value(), JsonNode.class);
+            return new CdcSourceRecord(null, keyNode, valueNode);
         } catch (Exception e) {
             LOG.error("Invalid Json:\n{}", new String(message.value()));
             throw e;