[Fix][Connector-V2] Fix assert connector assertion errors

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

@@ -49,7 +49,6 @@ public class AssertSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
     private static final Map<String, LongAccumulator> LONG_ACCUMULATOR = new ConcurrentHashMap<>();
     private static final Set<String> TABLE_NAMES = new CopyOnWriteArraySet<>();
     private final String catalogTableName;
-    private final long WAIT_SINK_WRITER_COMPLETE_TIME = 1000L;
 
     public AssertSinkWriter(
             SeaTunnelRowType seaTunnelRowType,
@@ -103,16 +102,14 @@ public void write(SeaTunnelRow element) {
 
     @Override
     public void close() {
-        try {
-            // When there are multiple AssertSinkWriters, some Sinks will run first, so let it wait
-            // for other Sinks, otherwise it will make incorrect judgments
-            Thread.sleep(WAIT_SINK_WRITER_COMPLETE_TIME);
-        } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-        }
         if (!assertRowRules.isEmpty()) {
             assertRowRules.entrySet().stream()
-                    .filter(entry -> !entry.getValue().isEmpty())
+                    .filter(
+                            entry ->
+                                    !entry.getValue().isEmpty()
+                                            && (assertRowRules.size() == 1
+                                                    || entry.getKey()
+                                                            .equals(this.catalogTableName)))
                     .forEach(
                             entry -> {
                                 List<AssertFieldRule.AssertRule> assertRules = entry.getValue();

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/java/org/apache/seatunnel/e2e/connector/assertion/FakeSourceToAssertIT.java

@@ -18,9 +18,7 @@
 package org.apache.seatunnel.e2e.connector.assertion;
 
 import org.apache.seatunnel.e2e.common.TestSuiteBase;
-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.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.TestTemplate;
@@ -47,10 +45,6 @@ public void testFakeSourceToAssertRowSink(TestContainer container)
     }
 
     @TestTemplate
-    @DisabledOnContainer(
-            value = {},
-            type = {EngineType.FLINK},
-            disabledReason = "Currently FLINK unsupported multi table")
     public void testFakeSourceToMultiAssertSink(TestContainer container)
             throws IOException, InterruptedException {
         Container.ExecResult execResult =
@@ -59,14 +53,18 @@ public void testFakeSourceToMultiAssertSink(TestContainer container)
     }
 
     @TestTemplate
-    @DisabledOnContainer(
-            value = {},
-            type = {EngineType.FLINK},
-            disabledReason = "Currently FLINK engine unsupported NULL type")
     public void testFakeFullTypesToAssertSink(TestContainer container)
             throws IOException, InterruptedException {
         Container.ExecResult execResult =
                 container.executeJob("/assertion/fake_full_types_to_assert.conf");
         Assertions.assertEquals(0, execResult.getExitCode());
     }
+
+    @TestTemplate
+    public void testFakeToAssertSinkWithError(TestContainer container)
+            throws IOException, InterruptedException {
+        Container.ExecResult execResult =
+                container.executeJob("/assertion/fake_full_types_to_assert.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+    }
 }

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/resources/assertion/fakesource_to_assert_with_error.conf

@@ -0,0 +1,160 @@
+#
+# 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 {
+  job.mode = "BATCH"
+}
+
+source {
+  FakeSource {
+    parallelism = 1
+    tables_configs = [
+      {
+        row.num = 100
+        schema = {
+          table = "test.abc"
+          columns = [
+            {
+              name = "id"
+              type = "bigint"
+            },
+            {
+              name = "name"
+              type = "string"
+            },
+            {
+              name = "age"
+              type = "int"
+            }
+          ]
+        }
+      },
+      {
+        row.num = 100
+        schema = {
+          table = "test.xyz"
+          columns = [
+            {
+              name = "id"
+              type = "bigint"
+            },
+            {
+              name = "name"
+              type = "string"
+            },
+            {
+              name = "age"
+              type = "int"
+            }
+          ]
+        }
+      },
+      {
+        row.num = 1000
+        schema = {
+          table = "test.www"
+          columns = [
+            {
+              name = "id"
+              type = "bigint"
+            },
+            {
+              name = "name"
+              type = "string"
+            },
+            {
+              name = "age"
+              type = "int"
+            }
+          ]
+        }
+      }
+    ]
+  }
+}
+
+transform {
+  FilterRowKind {
+    table_match_regex = "test.a.*"
+    table_transform = [{
+      table_path = "test.xyz"
+      exclude_kinds = ["INSERT"]
+    }]
+    exclude_kinds = ["INSERT"]
+  }
+}
+
+sink {
+  Assert {
+    rules =
+      {
+        tables_configs = [
+          {
+            table_path = "test.abc"
+            row_rules = [
+              {
+                rule_type = MIN_ROW
+                rule_value = 0
+              },
+              {
+                rule_type = MAX_ROW
+                rule_value = 0
+              }
+            ]
+          },
+          {
+            table_path = "test.xyz"
+            row_rules = [
+              {
+                rule_type = MIN_ROW
+                rule_value = 0
+              },
+              {
+                rule_type = MAX_ROW
+                rule_value = 0
+              }
+            ]
+          },
+          {
+            table_path = "test.www"
+            row_rules = [
+              {
+                rule_type = MIN_ROW
+                rule_value = 1000
+              },
+              {
+                rule_type = MAX_ROW
+                rule_value = 1000
+              }
+            ]
+            field_rules = [
+                {
+                field_name = name
+                field_type = string
+                field_value = [
+                    {
+                        rule_type = NOT_NULL
+                    }
+                ]
+              }
+            ]
+          }
+        ]
+      }
+  }
+}