(paimon) branch master updated: [Fix] MongoDB CDC: handle null fullDocument in transaction to prevent NullNode exception (#9332) (#9336)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new a51754c3f7 [Fix] MongoDB CDC: handle null fullDocument in transaction 
to prevent NullNode exception (#9332) (#9336)
a51754c3f7 is described below

commit a51754c3f7a1a16fcc343847410d6d065c9eb291
Author: Arvin <31638790+zhang-ar...@users.noreply.github.com>
AuthorDate: Sun Aug 30 10:01:38 2026 +0800

    [Fix] MongoDB CDC: handle null fullDocument in transaction to prevent 
NullNode exception (#9332) (#9336)
---
 .../action/cdc/mongodb/strategy/Mongo4VersionStrategy.java  | 13 +++++++++++--
 .../action/cdc/mongodb/strategy/MongoVersionStrategy.java   |  3 +++
 2 files changed, 14 insertions(+), 2 deletions(-)

diff --git 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/Mongo4VersionStrategy.java
 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/Mongo4VersionStrategy.java
index 5f9538d2fc..bbdc7a8d01 100644
--- 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/Mongo4VersionStrategy.java
+++ 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/Mongo4VersionStrategy.java
@@ -92,7 +92,10 @@ public class Mongo4VersionStrategy implements 
MongoVersionStrategy {
 
         switch (op) {
             case OP_INSERT:
-                records.add(processRecord(fullDocument, RowKind.INSERT));
+                RichCdcMultiplexRecord insertRecord = 
processRecord(fullDocument, RowKind.INSERT);
+                if (insertRecord != null) {
+                    records.add(insertRecord);
+                }
                 break;
             case OP_REPLACE:
             case OP_UPDATE:
@@ -100,7 +103,10 @@ public class Mongo4VersionStrategy implements 
MongoVersionStrategy {
                 // information. Therefore, data is first deleted using the 
primary key '_id', and
                 // then inserted.
                 records.add(processRecord(documentKey, RowKind.DELETE));
-                records.add(processRecord(fullDocument, RowKind.INSERT));
+                RichCdcMultiplexRecord updateRecord = 
processRecord(fullDocument, RowKind.INSERT);
+                if (updateRecord != null) {
+                    records.add(updateRecord);
+                }
                 break;
             case OP_DELETE:
                 records.add(processRecord(documentKey, RowKind.DELETE));
@@ -125,6 +131,9 @@ public class Mongo4VersionStrategy implements 
MongoVersionStrategy {
         CdcSchema.Builder schemaBuilder = CdcSchema.newBuilder();
         Map<String, String> record =
                 getExtractRow(fullDocument, schemaBuilder, computedColumns, 
mongodbConfig);
+        if (record == null) {
+            return null;
+        }
         schemaBuilder.primaryKey(extractPrimaryKeys());
         return new RichCdcMultiplexRecord(
                 databaseName, collection, schemaBuilder.build(), new 
CdcRecord(rowKind, record));
diff --git 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/MongoVersionStrategy.java
 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/MongoVersionStrategy.java
index a41a1eec68..fcabcaffbb 100644
--- 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/MongoVersionStrategy.java
+++ 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/strategy/MongoVersionStrategy.java
@@ -82,6 +82,9 @@ public interface MongoVersionStrategy {
             List<ComputedColumn> computedColumns,
             Configuration mongodbConfig)
             throws JsonProcessingException {
+        if (jsonNode == null || jsonNode.isNull()) {
+            return null;
+        }
         SchemaAcquisitionMode mode =
                 
SchemaAcquisitionMode.valueOf(mongodbConfig.get(START_MODE).toUpperCase());
         ObjectNode objectNode =