[Improve][Core][Exceptiom Management] Add Exception Management API & Unified Exception in API Module

seatunnel-api/src/main/java/org/apache/seatunnel/api/common/PrepareFailException.java

@@ -18,21 +18,22 @@
 package org.apache.seatunnel.api.common;
 
 import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
 /**
  * This exception will throw when {@link SeaTunnelPluginLifeCycle#prepare(Config)} failed.
  */
-public class PrepareFailException extends RuntimeException {
+public class PrepareFailException extends SeaTunnelRuntimeException {
 
     public PrepareFailException(String pluginName, PluginType type, String message) {
-        super(String.format("PluginName: %s, PluginType: %s, Message: %s", pluginName, type.getType(),
-                message));
+        super(SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED, String.format("PluginName: %s, PluginType: %s, Message: %s",
+                pluginName, type.getType(), message));
     }
 
     public PrepareFailException(String pluginName, PluginType type, String message, Throwable cause) {
-        super(String.format("PluginName: %s, PluginType: %s, Message: %s", pluginName, type.getType(),
-                message), cause);
+        super(SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED, String.format("PluginName: %s, PluginType: %s, Message: %s",
+                pluginName, type.getType(), message), cause);
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/common/SeaTunnelAPIErrorCode.java

@@ -0,0 +1,47 @@
+/*
+ * 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.seatunnel.api.common;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+
+public enum SeaTunnelAPIErrorCode implements SeaTunnelErrorCode {
+    CONFIG_VALIDATION_FAILED("API-01", "Configuration item validate failed"),
+    OPTION_VALIDATION_FAILED("API-02", "Option item validate failed"),
+    CATALOG_INITIALIZE_FAILED("API-03", "Catalog initialize failed"),
+    DATABASE_NOT_EXISTED("API-04", "Database not existed"),
+    TABLE_NOT_EXISTED("API-05", "Table not existed"),
+    FACTORY_INITIALIZE_FAILED("API-06", "Factory initialize failed"),;
+
+    private final String code;
+    private final String description;
+
+    SeaTunnelAPIErrorCode(String code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+
+    @Override
+    public String getCode() {
+        return code;
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+}

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/util/OptionValidationException.java

@@ -17,16 +17,19 @@
 
 package org.apache.seatunnel.api.configuration.util;
 
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
 /**
  * Exception for all errors occurring during option validation phase.
  */
-public class OptionValidationException extends RuntimeException {
+public class OptionValidationException extends SeaTunnelRuntimeException {
 
     public OptionValidationException(String message, Throwable cause) {
-        super(message, cause);
+        super(SeaTunnelAPIErrorCode.OPTION_VALIDATION_FAILED, message, cause);
     }
 
     public OptionValidationException(String message) {
-        super(message);
+        super(SeaTunnelAPIErrorCode.OPTION_VALIDATION_FAILED, message);
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/exception/CatalogException.java

@@ -17,30 +17,33 @@
 
 package org.apache.seatunnel.api.table.catalog.exception;
 
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
 /**
  * A catalog-related, runtime exception.
  */
-public class CatalogException extends RuntimeException {
+public class CatalogException extends SeaTunnelRuntimeException {
 
     /**
      * @param message the detail message.
      */
     public CatalogException(String message) {
-        super(message);
+        super(SeaTunnelAPIErrorCode.CATALOG_INITIALIZE_FAILED, message);
     }
 
     /**
      * @param cause the cause.
      */
     public CatalogException(Throwable cause) {
-        super(cause);
+        super(SeaTunnelAPIErrorCode.CATALOG_INITIALIZE_FAILED, cause);
     }
 
     /**
      * @param message the detail message.
      * @param cause   the cause.
      */
     public CatalogException(String message, Throwable cause) {
-        super(message, cause);
+        super(SeaTunnelAPIErrorCode.CATALOG_INITIALIZE_FAILED, message, cause);
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/exception/DatabaseNotExistException.java

@@ -18,12 +18,15 @@
 
 package org.apache.seatunnel.api.table.catalog.exception;
 
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
 /** Exception for trying to operate on a database that doesn't exist. */
-public class DatabaseNotExistException extends Exception {
+public class DatabaseNotExistException extends SeaTunnelRuntimeException {
     private static final String MSG = "Database %s does not exist in Catalog %s.";
 
     public DatabaseNotExistException(String catalogName, String databaseName, Throwable cause) {
-        super(String.format(MSG, databaseName, catalogName), cause);
+        super(SeaTunnelAPIErrorCode.DATABASE_NOT_EXISTED, String.format(MSG, databaseName, catalogName), cause);
     }
 
     public DatabaseNotExistException(String catalogName, String databaseName) {

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/exception/TableNotExistException.java

@@ -18,10 +18,12 @@
 
 package org.apache.seatunnel.api.table.catalog.exception;
 
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
 import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
 
 /** Exception for trying to operate on a table that doesn't exist. */
-public class TableNotExistException extends Exception {
+public class TableNotExistException extends SeaTunnelRuntimeException {
 
     private static final String MSG = "Table %s does not exist in Catalog %s.";
 
@@ -30,6 +32,6 @@ public TableNotExistException(String catalogName, TablePath tablePath) {
     }
 
     public TableNotExistException(String catalogName, TablePath tablePath, Throwable cause) {
-        super(String.format(MSG, tablePath.getFullName(), catalogName), cause);
+        super(SeaTunnelAPIErrorCode.TABLE_NOT_EXISTED, String.format(MSG, tablePath.getFullName(), catalogName), cause);
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryException.java

@@ -17,13 +17,16 @@
 
 package org.apache.seatunnel.api.table.factory;
 
-public class FactoryException extends RuntimeException {
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+public class FactoryException extends SeaTunnelRuntimeException {
 
     public FactoryException(String message, Throwable cause) {
-        super(message, cause);
+        super(SeaTunnelAPIErrorCode.FACTORY_INITIALIZE_FAILED, message, cause);
     }
 
     public FactoryException(String message) {
-        super(message);
+        super(SeaTunnelAPIErrorCode.FACTORY_INITIALIZE_FAILED, message);
     }
 }

seatunnel-common/src/main/java/org/apache/seatunnel/common/exception/CommonErrorCode.java

@@ -0,0 +1,52 @@
+/*
+ * 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.seatunnel.common.exception;
+
+public enum CommonErrorCode implements SeaTunnelErrorCode {
+    FILE_OPERATION_FAILED("COMMON-01", "File operation failed, such as (read,list,write,move,copy,sync) etc..."),
+    JSON_OPERATION_FAILED("COMMON-02", "Json covert/parse operation failed"),
+    REFLECT_CLASS_OPERATION_FAILED("COMMON-03", "Reflect class operation failed"),
+    SERIALIZE_OPERATION_FAILED("COMMON-04", "Serialize class operation failed"),
+    UNSUPPORTED_OPERATION("COMMON-05", "Unsupported operation"),
+    ILLEGAL_ARGUMENT("COMMON-06", "Illegal argument"),
+    UNSUPPORTED_DATA_TYPE("COMMON-07", "Unsupported data type"),
+    SQL_OPERATION_FAILED("COMMON-08", "Sql operation failed, such as (execute,addBatch,close) etc..."),
+    TABLE_SCHEMA_GET_FAILED("COMMON-09", "Get table schema from upstream data failed"),
+    FLUSH_DATA_FAILED("COMMON-10", "Flush data operation that in sink connector failed"),
+    WRITER_OPERATION_FAILED("COMMON-11", "Sink writer operation failed, such as (open, close) etc..."),
+    READER_OPERATION_FAILED("COMMON-12", "Source reader operation failed, such as (open, close) etc..."),
+    HTTP_OPERATION_FAILED("COMMON-13", "Http operation failed, such as (open, close, response) etc...");
+
+    private final String code;
+    private final String description;
+
+    CommonErrorCode(String code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+
+    @Override
+    public String getCode() {
+        return code;
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+}

seatunnel-common/src/main/java/org/apache/seatunnel/common/exception/SeaTunnelErrorCode.java

@@ -0,0 +1,39 @@
+/*
+ * 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.seatunnel.common.exception;
+
+/**
+ * SeaTunnel connector error code interface
+ */
+public interface SeaTunnelErrorCode {
+    /**
+     * Get error code
+     * @return error code
+     */
+    String getCode();
+
+    /**
+     * Get error description
+     * @return error description
+     */
+    String getDescription();
+
+    default String getErrorMessage() {
+        return String.format("ErrorCode:[%s], ErrorDescription:[%s]", getCode(), getDescription());
+    }
+}

seatunnel-common/src/main/java/org/apache/seatunnel/common/exception/SeaTunnelRuntimeException.java

@@ -0,0 +1,53 @@
+/*
+ * 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.seatunnel.common.exception;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+/**
+ * SeaTunnel global exception, used to tell user more clearly error messages
+ */
+public class SeaTunnelRuntimeException extends RuntimeException {
+    private final SeaTunnelErrorCode seaTunnelErrorCode;
+
+    public SeaTunnelRuntimeException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage) {
+        super(seaTunnelErrorCode.getErrorMessage() + " - " + errorMessage);
+        this.seaTunnelErrorCode = seaTunnelErrorCode;
+    }
+
+    public SeaTunnelRuntimeException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage, Throwable cause) {
+        super(seaTunnelErrorCode.getErrorMessage() + " - " + errorMessage, cause);
+        this.seaTunnelErrorCode = seaTunnelErrorCode;
+    }
+
+    public SeaTunnelRuntimeException(SeaTunnelErrorCode seaTunnelErrorCode, Throwable cause) {
+        super(seaTunnelErrorCode.getErrorMessage() + " - " + getMessageFromThrowable(cause));
+        this.seaTunnelErrorCode = seaTunnelErrorCode;
+    }
+
+    public static String getMessageFromThrowable(Throwable cause) {
+        if (cause == null) {
+            return "";
+        }
+        StringWriter stringWriter = new StringWriter();
+        PrintWriter printWriter = new PrintWriter(stringWriter);
+        cause.printStackTrace(printWriter);
+        return stringWriter.toString();
+    }
+}