[flink] Supports debezium-bson format of kafka which collected from mongodb via debezium

paimon-flink/paimon-flink-cdc/pom.xml

@@ -44,6 +44,7 @@ under the License.
         <flink.connector.pulsar.version>4.0.0-1.17</flink.connector.pulsar.version>
         <confluent.platform.version>7.5.0</confluent.platform.version>
         <flink.connector.kafka.version>3.3.0-1.20</flink.connector.kafka.version>
+        <bson.version>5.2.1</bson.version>
     </properties>
 
     <repositories>
@@ -165,6 +166,13 @@ under the License.
             <version>${json-path.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.mongodb</groupId>
+            <artifactId>bson</artifactId>
+            <version>${bson.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
         <!-- test dependencies -->
 
         <dependency>

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

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.format.debezium;
+
+import org.apache.paimon.flink.action.cdc.format.AbstractJsonDataFormat;
+import org.apache.paimon.flink.action.cdc.format.RecordParserFactory;
+
+/**
+ * Supports the message queue's debezium bson data format and provides definitions for the message
+ * queue's record json deserialization class and parsing class {@link DebeziumBsonRecordParser}.
+ */
+public class DebeziumBsonDataFormat extends AbstractJsonDataFormat {
+
+    @Override
+    protected RecordParserFactory parser() {
+        return DebeziumBsonRecordParser::new;
+    }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.format.debezium;
+
+import org.apache.paimon.flink.action.cdc.format.DataFormat;
+import org.apache.paimon.flink.action.cdc.format.DataFormatFactory;
+
+/** Factory to create {@link DebeziumBsonDataFormat}. */
+public class DebeziumBsonDataFormatFactory implements DataFormatFactory {
+
+    public static final String IDENTIFIER = "debezium-bson";
+
+    @Override
+    public String identifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public DataFormat create() {
+        return new DebeziumBsonDataFormat();
+    }
+}

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

@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action.cdc.format.debezium;
+
+import org.apache.paimon.flink.action.cdc.CdcSourceRecord;
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
+import org.apache.paimon.flink.action.cdc.mongodb.BsonValueConvertor;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.TypeUtils;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import org.bson.BsonDocument;
+import org.bson.BsonValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+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.utils.JsonSerdeUtil.writeValueAsString;
+
+/**
+ * The {@code DebeziumRecordParser} class extends the {@link DebeziumJsonRecordParser} and is
+ * designed to parse records from MongoDB's BSON change data capture (CDC) format via Debezium.
+ *
+ * <p>The class supports The following features:
+ *
+ * <p>Convert bson string to java object from before/after field
+ *
+ * <p>Parse the schema from bson, all fields are string type, and the _id field is the primary key
+ */
+public class DebeziumBsonRecordParser extends DebeziumJsonRecordParser {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DebeziumBsonRecordParser.class);
+
+    private static final String FIELD_COLLECTION = "collection";
+    private static final String FIELD_OBJECT_ID = "_id";
+    private static final List<String> PRIMARY_KEYS = Collections.singletonList(FIELD_OBJECT_ID);
+
+    public DebeziumBsonRecordParser(TypeMapping typeMapping, List<ComputedColumn> computedColumns) {
+        super(typeMapping, computedColumns);
+    }
+
+    @Override
+    protected void setRoot(CdcSourceRecord record) {
+        JsonNode node = (JsonNode) record.getValue();
+        if (node.has(FIELD_SCHEMA)) {
+            root = node.get(FIELD_PAYLOAD);
+        } else {
+            root = node;
+        }
+    }
+
+    @Override
+    protected Map<String, String> extractRowData(JsonNode record, RowType.Builder rowTypeBuilder) {
+        // bson record should be a string
+        Preconditions.checkArgument(
+                record.isTextual(),
+                "debezium bson record expected to be STRING type, but actual is %s",
+                record.getNodeType());
+
+        BsonDocument document = BsonDocument.parse(record.asText());
+        LinkedHashMap<String, String> resultMap = new LinkedHashMap<>();
+        for (Map.Entry<String, BsonValue> entry : document.entrySet()) {
+            String fieldName = entry.getKey();
+            resultMap.put(fieldName, toJsonString(BsonValueConvertor.convert(entry.getValue())));
+            rowTypeBuilder.field(fieldName, DataTypes.STRING());
+        }
+
+        evalComputedColumns(resultMap, rowTypeBuilder);
+
+        return resultMap;
+    }
+
+    private static String toJsonString(Object entry) {
+        if (entry == null) {
+            return null;
+        } else if (!TypeUtils.isBasicType(entry)) {
+            try {
+                return writeValueAsString(entry);
+            } catch (JsonProcessingException e) {
+                LOG.error("Failed to deserialize record.", e);
+            }
+        }
+        return Objects.toString(entry);
+    }
+
+    @Override
+    protected List<String> extractPrimaryKeys() {
+        return PRIMARY_KEYS;
+    }
+
+    @Nullable
+    @Override
+    protected String getTableName() {
+        return getFromSourceField(FIELD_COLLECTION);
+    }
+
+    @Override
+    protected String format() {
+        return "debezium-bson";
+    }
+}

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

@@ -244,7 +244,7 @@ protected String format() {
     }
 
     @Nullable
-    private String getFromSourceField(String key) {
+    protected String getFromSourceField(String key) {
         JsonNode node = root.get(FIELD_SOURCE);
         if (isNull(node)) {
             return null;