[Improve][Connector-V2][MongoDB] Unified exception for MongoDB source & sink connector

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/data/DataTypeValidator.java

@@ -21,17 +21,21 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.api.table.type.SqlType;
+import org.apache.seatunnel.common.exception.CommonErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.mongodb.exception.MongodbConnectorException;
 
 public class DataTypeValidator {
 
     public static void validateDataType(SeaTunnelDataType dataType) throws IllegalArgumentException {
         switch (dataType.getSqlType()) {
             case TIME:
-                throw new IllegalArgumentException("Unsupported data type: " + dataType);
+                throw new MongodbConnectorException(CommonErrorCode.UNSUPPORTED_DATA_TYPE,
+                    "Unsupported data type: " + dataType);
             case MAP:
                 MapType mapType = (MapType) dataType;
                 if (!SqlType.STRING.equals(mapType.getKeyType().getSqlType())) {
-                    throw new IllegalArgumentException("Unsupported map key type: " + mapType.getKeyType());
+                    throw new MongodbConnectorException(CommonErrorCode.UNSUPPORTED_DATA_TYPE,
+                        "Unsupported map key type: " + mapType.getKeyType());
                 }
                 break;
             case ROW:

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/data/DefaultSerializer.java

@@ -47,6 +47,7 @@ public DefaultSerializer(@NonNull SeaTunnelRowType rowType) {
         this.rowType = rowType;
     }
 
+    @Override
     public Document serialize(@NonNull SeaTunnelRow row) {
         return convert(rowType, row);
     }

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/data/SeaTunnelRowBsonWriter.java

@@ -22,6 +22,8 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.exception.CommonErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.mongodb.exception.MongodbConnectorException;
 
 import lombok.Getter;
 import lombok.SneakyThrows;
@@ -225,27 +227,32 @@ protected void doWriteTimestamp(BsonTimestamp value) {
 
     @Override
     protected void doWriteJavaScriptWithScope(String value) {
-        throw new UnsupportedOperationException("Unsupported JavaScriptWithScope");
+        throw new MongodbConnectorException(CommonErrorCode.UNSUPPORTED_OPERATION,
+            "Unsupported JavaScriptWithScope");
     }
 
     @Override
     protected void doWriteMaxKey() {
-        throw new UnsupportedOperationException("Unsupported MaxKey");
+        throw new MongodbConnectorException(CommonErrorCode.UNSUPPORTED_OPERATION,
+            "Unsupported MaxKey");
     }
 
     @Override
     protected void doWriteMinKey() {
-        throw new UnsupportedOperationException("Unsupported MinKey");
+        throw new MongodbConnectorException(CommonErrorCode.UNSUPPORTED_OPERATION,
+            "Unsupported MinKey");
     }
 
     @Override
     protected void doWriteRegularExpression(BsonRegularExpression value) {
-        throw new UnsupportedOperationException("Unsupported BsonRegularExpression");
+        throw new MongodbConnectorException(CommonErrorCode.UNSUPPORTED_OPERATION,
+            "Unsupported BsonRegularExpression");
     }
 
     @Override
     protected void doWriteDBPointer(BsonDbPointer value) {
-        throw new UnsupportedOperationException("Unsupported BsonDbPointer");
+        throw new MongodbConnectorException(CommonErrorCode.UNSUPPORTED_OPERATION,
+            "Unsupported BsonDbPointer");
     }
 
     @Override

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/exception/MongodbConnectorException.java

@@ -0,0 +1,36 @@
+/*
+ * 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
+ *
+ *    https://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.seatunnel.connectors.seatunnel.mongodb.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+public class MongodbConnectorException extends SeaTunnelRuntimeException {
+
+    public MongodbConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage) {
+        super(seaTunnelErrorCode, errorMessage);
+    }
+
+    public MongodbConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage, Throwable cause) {
+        super(seaTunnelErrorCode, errorMessage, cause);
+    }
+
+    public MongodbConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, Throwable cause) {
+        super(seaTunnelErrorCode, cause);
+    }
+}

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbSink.java

@@ -22,6 +22,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.mongodb.config.MongodbConfig.URI;
 
 import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
@@ -34,6 +35,7 @@
 import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
 import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
 import org.apache.seatunnel.connectors.seatunnel.mongodb.config.MongodbParameters;
+import org.apache.seatunnel.connectors.seatunnel.mongodb.exception.MongodbConnectorException;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigBeanFactory;
@@ -58,7 +60,9 @@ public String getPluginName() {
     public void prepare(Config config) throws PrepareFailException {
         CheckResult result = CheckConfigUtil.checkAllExists(config, URI.key(), DATABASE.key(), COLLECTION.key());
         if (!result.isSuccess()) {
-            throw new PrepareFailException(getPluginName(), PluginType.SOURCE, result.getMsg());
+            throw new MongodbConnectorException(SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+                String.format("PluginName: %s, PluginType: %s, Message: %s",
+                    getPluginName(), PluginType.SINK, result.getMsg()));
         }
 
         this.params = ConfigBeanFactory.create(config, MongodbParameters.class);

seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/source/MongodbSource.java

@@ -22,6 +22,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.mongodb.config.MongodbConfig.URI;
 
 import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
 import org.apache.seatunnel.api.source.Boundedness;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
@@ -35,6 +36,7 @@
 import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitSource;
 import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
 import org.apache.seatunnel.connectors.seatunnel.mongodb.config.MongodbParameters;
+import org.apache.seatunnel.connectors.seatunnel.mongodb.exception.MongodbConnectorException;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigBeanFactory;
@@ -57,7 +59,9 @@ public String getPluginName() {
     public void prepare(Config config) throws PrepareFailException {
         CheckResult result = CheckConfigUtil.checkAllExists(config, URI.key(), DATABASE.key(), COLLECTION.key());
         if (!result.isSuccess()) {
-            throw new PrepareFailException(getPluginName(), PluginType.SOURCE, result.getMsg());
+            throw new MongodbConnectorException(SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+                String.format("PluginName: %s, PluginType: %s, Message: %s",
+                    getPluginName(), PluginType.SOURCE, result.getMsg()));
         }
 
         this.params = ConfigBeanFactory.create(config, MongodbParameters.class);