[Feature][Connector] add IT for Assert Sink in e2e module

seatunnel-connectors/plugin-mapping.properties

@@ -41,6 +41,7 @@ flink.sink.FileSink = seatunnel-connector-flink-file
 flink.sink.InfluxDbSink = seatunnel-connector-flink-influxdb
 flink.sink.JdbcSink = seatunnel-connector-flink-jdbc
 flink.sink.Kafka = seatunnel-connector-flink-kafka
+flink.sink.AssertSink = seatunnel-connector-flink-assert
 
 # Spark Source
 

seatunnel-e2e/seatunnel-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/assertion/FakeSourceToAssertIT.java

@@ -0,0 +1,35 @@
+/*
+ * 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.e2e.flink.assertion;
+
+import org.apache.seatunnel.e2e.flink.FlinkContainer;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.testcontainers.containers.Container;
+
+import java.io.IOException;
+
+public class FakeSourceToAssertIT extends FlinkContainer {
+
+    @Test
+    public void testFakeSourceToConsoleSink() throws IOException, InterruptedException {
+        Container.ExecResult execResult = executeSeaTunnelFlinkJob("/assertion/fakesource_to_assert.conf");
+        Assert.assertEquals(0, execResult.getExitCode());
+    }
+}

seatunnel-e2e/seatunnel-flink-e2e/src/test/resources/assertion/fakesource_to_assert.conf

@@ -0,0 +1,88 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  # You can set flink configuration here
+  execution.parallelism = 1
+  #execution.checkpoint.interval = 10000
+  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+  # This is a example source plugin **only for test and demonstrate the feature source plugin**
+    FakeSource {
+      result_table_name = "fake"
+      field_name = "name,age"
+    }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+}
+
+transform {
+    sql {
+      sql = "select name,age from fake"
+    }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}
+
+sink {
+   AssertSink {
+       rules = 
+           [{
+               field_name = name
+               field_type = string
+               field_value = [
+                   {
+                       rule_type = NOT_NULL
+                   },
+                   {
+                       rule_type = MIN_LENGTH
+                       rule_value = 3
+                   },
+                   {
+                        rule_type = MAX_LENGTH
+                        rule_value = 5
+                   }
+               ]
+           },{
+               field_name = age
+               field_type = int
+               field_value = [
+                   {
+                       rule_type = NOT_NULL
+                   },
+                   {
+                       rule_type = MIN
+                       rule_value = 10
+                   },
+                   {
+                        rule_type = MAX
+                        rule_value = 20
+                   }
+               ]
+           }
+           ]
+   }
+  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
+}