[bugfix][connector-mongodb] fix mongodb null value write

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/serde/RowDataToBsonConverters.java

@@ -92,11 +92,7 @@ private static SerializableFunction<Object, BsonValue> wrapIntoNullSafeInternalC
             @Override
             public BsonValue apply(Object value) {
                 if (value == null || NULL.equals(type.getSqlType())) {
-                    throw new MongodbConnectorException(
-                            UNSUPPORTED_DATA_TYPE,
-                            "The column type is <"
-                                    + type
-                                    + ">, but a null value is being written into it");
+                    return new BsonNull();
                 } else {
                     return internalConverter.apply(value);
                 }

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/AbstractMongodbIT.java

@@ -58,6 +58,8 @@ public abstract class AbstractMongodbIT extends TestSuiteBase implements TestRes
 
     protected static final List<Document> TEST_SPLIT_DATASET = generateTestDataSet(10);
 
+    protected static final List<Document> TEST_NULL_DATASET = generateTestDataSetWithNull(10);
+
     protected static final String MONGODB_IMAGE = "mongo:latest";
 
     protected static final String MONGODB_CONTAINER_HOST = "e2e_mongodb";
@@ -70,6 +72,10 @@ public abstract class AbstractMongodbIT extends TestSuiteBase implements TestRes
 
     protected static final String MONGODB_SPLIT_TABLE = "test_split_op_db";
 
+    protected static final String MONGODB_NULL_TABLE = "test_null_op_db";
+
+    protected static final String MONGODB_NULL_TABLE_RESULT = "test_null_op_db_result";
+
     protected static final String MONGODB_MATCH_RESULT_TABLE = "test_match_op_result_db";
 
     protected static final String MONGODB_SPLIT_RESULT_TABLE = "test_split_op_result_db";
@@ -101,15 +107,18 @@ public void initConnection() {
     protected void initSourceData() {
         MongoCollection<Document> sourceMatchTable =
                 client.getDatabase(MONGODB_DATABASE).getCollection(MONGODB_MATCH_TABLE);
-
         sourceMatchTable.deleteMany(new Document());
         sourceMatchTable.insertMany(TEST_MATCH_DATASET);
 
         MongoCollection<Document> sourceSplitTable =
                 client.getDatabase(MONGODB_DATABASE).getCollection(MONGODB_SPLIT_TABLE);
-
         sourceSplitTable.deleteMany(new Document());
         sourceSplitTable.insertMany(TEST_SPLIT_DATASET);
+
+        MongoCollection<Document> sourceNullTable =
+                client.getDatabase(MONGODB_DATABASE).getCollection(MONGODB_NULL_TABLE);
+        sourceNullTable.deleteMany(new Document());
+        sourceNullTable.insertMany(TEST_NULL_DATASET);
     }
 
     protected void clearDate(String table) {
@@ -169,6 +178,23 @@ public static List<Document> generateTestDataSet(int count) {
         return dataSet;
     }
 
+    public static List<Document> generateTestDataSetWithNull(int count) {
+        List<Document> dataSet = new ArrayList<>();
+
+        for (int i = 0; i < count; i++) {
+            dataSet.add(
+                    new Document("c_map", null)
+                            .append("c_array", null)
+                            .append("c_string", null)
+                            .append("c_boolean", null)
+                            .append("c_int", null)
+                            .append("c_bigint", null)
+                            .append("c_double", null)
+                            .append("c_row", null));
+        }
+        return dataSet;
+    }
+
     protected static String randomString() {
         int length = RANDOM.nextInt(10) + 1;
         StringBuilder sb = new StringBuilder(length);

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java

@@ -17,7 +17,9 @@
 
 package org.apache.seatunnel.e2e.connector.v2.mongodb;
 
+import org.apache.seatunnel.e2e.common.container.EngineType;
 import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
 
 import org.bson.Document;
 import org.junit.jupiter.api.Assertions;
@@ -43,6 +45,24 @@ public void testMongodbSourceAndSink(TestContainer container)
         clearDate(MONGODB_SINK_TABLE);
     }
 
+    @TestTemplate
+    @DisabledOnContainer(
+            value = {},
+            type = {EngineType.FLINK, EngineType.SPARK},
+            disabledReason = "Currently SPARK and FLINK do not support mongodb null value write")
+    public void testMongodbNullValue(TestContainer container)
+            throws IOException, InterruptedException {
+        Container.ExecResult nullResult = container.executeJob("/mongodb_null_value.conf");
+        Assertions.assertEquals(0, nullResult.getExitCode(), nullResult.getStderr());
+        Assertions.assertIterableEquals(
+                TEST_NULL_DATASET.stream().peek(e -> e.remove("_id")).collect(Collectors.toList()),
+                readMongodbData(MONGODB_NULL_TABLE_RESULT).stream()
+                        .peek(e -> e.remove("_id"))
+                        .collect(Collectors.toList()));
+        clearDate(MONGODB_NULL_TABLE);
+        clearDate(MONGODB_NULL_TABLE_RESULT);
+    }
+
     @TestTemplate
     public void testMongodbSourceMatch(TestContainer container)
             throws IOException, InterruptedException {

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_null_value.conf

@@ -0,0 +1,89 @@
+#
+# 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.
+#
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+  #spark config
+  spark.app.name = "SeaTunnel"
+  spark.executor.instances = 1
+  spark.executor.cores = 1
+  spark.executor.memory = "1g"
+  spark.master = local
+}
+
+source {
+  MongoDB {
+    uri = "mongodb://e2e_mongodb:27017/test_db"
+    database = "test_db"
+    collection = "test_null_op_db"
+    match.projection = "{ c_bigint:0 }"
+    result_table_name = "mongodb_null_table"
+    cursor.no-timeout = true
+    fetch.size = 1000
+    max.time-min = 100
+    schema = {
+      fields {
+        c_map = "map<string, string>"
+        c_array = "array<int>"
+        c_string = string
+        c_boolean = boolean
+        c_int = int
+        c_bigint = bigint
+        c_double = double
+        c_row = {
+          c_map = "map<string, string>"
+          c_array = "array<int>"
+          c_string = string
+          c_boolean = boolean
+          c_int = int
+          c_bigint = bigint
+          c_double = double
+        }
+      }
+    }
+  }
+}
+
+sink {
+  MongoDB {
+    uri = "mongodb://e2e_mongodb:27017/test_db?retryWrites=true"
+    database = "test_db"
+    collection = "test_null_op_db_result"
+    schema = {
+      fields {
+        c_map = "map<string, string>"
+        c_array = "array<int>"
+        c_string = string
+        c_boolean = boolean
+        c_int = int
+        c_bigint = bigint
+        c_double = double
+        c_row = {
+          c_map = "map<string, string>"
+          c_array = "array<int>"
+          c_string = string
+          c_boolean = boolean
+          c_int = int
+          c_bigint = bigint
+          c_double = double
+        }
+      }
+    }
+  }
+
+}