[Improve][Connector-V2][Assert] Unified exception for assert connector (#3331)

* [Improve][Core][Exception Management] Add exception api

* remove duplicate ExceptionUtil class (#3037)

* [Improve][Core][Exception Management] Add exception api

* [Improve][Core][Exception Management] Rename exception name

* [Improve][Connector-V2][Assert] Unified exception for assert sink connector

* [Improve][Connector-V2][Assert] Fix code style

Co-authored-by: Eric <gaojun2048@gmail.com>

Author: TyrantLucifer

seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/exception/AssertConnectorErrorCode.java

@@ -0,0 +1,42 @@
+/*
+ * 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.connectors.seatunnel.assertion.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+
+public enum AssertConnectorErrorCode implements SeaTunnelErrorCode {
+    RULE_VALIDATION_FAILED("ASSERT-01", "Rule validate failed");
+
+    private final String code;
+    private final String description;
+
+    AssertConnectorErrorCode(String code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+
+    @Override
+    public String getCode() {
+        return code;
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+}

seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/exception/AssertConnectorException.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
+ *
+ *    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.connectors.seatunnel.assertion.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+public class AssertConnectorException extends SeaTunnelRuntimeException {
+
+    public AssertConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage) {
+        super(seaTunnelErrorCode, errorMessage);
+    }
+
+    public AssertConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage, Throwable cause) {
+        super(seaTunnelErrorCode, errorMessage, cause);
+    }
+
+    public AssertConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, Throwable cause) {
+        super(seaTunnelErrorCode, cause);
+    }
+}

seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkWriter.java

@@ -20,6 +20,8 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.connectors.seatunnel.assertion.excecutor.AssertExecutor;
+import org.apache.seatunnel.connectors.seatunnel.assertion.exception.AssertConnectorErrorCode;
+import org.apache.seatunnel.connectors.seatunnel.assertion.exception.AssertConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.assertion.rule.AssertFieldRule;
 import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
 
@@ -49,7 +51,8 @@ public void write(SeaTunnelRow element) {
             ASSERT_EXECUTOR
                 .fail(element, seaTunnelRowType, assertFieldRules)
                 .ifPresent(failRule -> {
-                    throw new IllegalStateException("row :" + element + " fail rule: " + failRule);
+                    throw new AssertConnectorException(AssertConnectorErrorCode.RULE_VALIDATION_FAILED,
+                            "row :" + element + " fail rule: " + failRule);
                 });
         }
     }
@@ -67,7 +70,8 @@ public void close() {
                         return false;
                 }
             }).findFirst().ifPresent(failRule -> {
-                throw new IllegalStateException("row num :" + LONG_ACCUMULATOR.longValue() + " fail rule: " + failRule);
+                throw new AssertConnectorException(AssertConnectorErrorCode.RULE_VALIDATION_FAILED,
+                        "row num :" + LONG_ACCUMULATOR.longValue() + " fail rule: " + failRule);
             });
         }
     }