[improve] add assert options (#8620)

Author: liunaijie

seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java

@@ -205,7 +205,6 @@ private Set<String> buildWhiteList() {
         whiteList.add("RocketMqSinkOptions");
         whiteList.add("ClickhouseFileSinkOptions");
         whiteList.add("IcebergSinkOptions");
-        whiteList.add("AssertSinkOptions");
         whiteList.add("MaxcomputeSourceOptions");
         whiteList.add("InfluxDBSourceOptions");
         whiteList.add("InfluxDBSinkOptions");

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

@@ -17,13 +17,6 @@
 
 package org.apache.seatunnel.connectors.seatunnel.assertion.sink;
 
-import org.apache.seatunnel.shade.com.fasterxml.jackson.core.type.TypeReference;
-
-import org.apache.seatunnel.api.configuration.Option;
-import org.apache.seatunnel.api.configuration.Options;
-
-import java.util.Map;
-
 public class AssertConfig {
 
     public static final String RULE_TYPE = "rule_type";
@@ -64,29 +57,12 @@ public class AssertConfig {
     public static final String COLUMN_DEFAULT_VALUE = "default_value";
     public static final String COLUMN_COMMENT = "comment";
 
+    public static final String TABLE_PATH = "table_path";
+
     public static class TableIdentifierRule {
         public static final String TABLE_IDENTIFIER_RULE = "table_identifier_rule";
 
         public static final String TABLE_IDENTIFIER_CATALOG_NAME = "catalog_name";
         public static final String TABLE_IDENTIFIER_TABLE_NAME = "table";
     }
-
-    public static final Option<String> COMMENT =
-            Options.key("comment")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("SeaTunnel Schema Column Comment");
-
-    public static final Option<Map<String, Object>> RULES =
-            Options.key("rules")
-                    .type(new TypeReference<Map<String, Object>>() {})
-                    .noDefaultValue()
-                    .withDescription(
-                            "Rule definition of user's available data. Each rule represents one field validation or row num validation.");
-
-    public static final Option<String> TABLE_PATH =
-            Options.key("table_path")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("table full path");
 }

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

@@ -45,8 +45,8 @@
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.CATALOG_TABLE_RULES;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.FIELD_RULES;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.ROW_RULES;
-import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.RULES;
 import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.TABLE_PATH;
+import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertSinkOptions.RULES;
 
 public class AssertSink extends AbstractSimpleSink<SeaTunnelRow, Void>
         implements SupportMultiTableSink {
@@ -71,7 +71,7 @@ public AssertSink(ReadonlyConfig pluginConfig, CatalogTable catalogTable) {
         if (ruleConfig.hasPath(TABLE_CONFIGS.key())) {
             List<? extends Config> tableConfigs = ruleConfig.getConfigList(TABLE_CONFIGS.key());
             for (Config tableConfig : tableConfigs) {
-                String tableName = tableConfig.getString(TABLE_PATH.key());
+                String tableName = tableConfig.getString(TABLE_PATH);
                 initTableRule(catalogTable, tableConfig, tableName);
             }
         } else {

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

@@ -18,15 +18,15 @@
 package org.apache.seatunnel.connectors.seatunnel.assertion.sink;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
-import org.apache.seatunnel.api.sink.SinkCommonOptions;
 import org.apache.seatunnel.api.table.connector.TableSink;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
 
 import com.google.auto.service.AutoService;
 
-import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertConfig.RULES;
+import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertSinkOptions.MULTI_TABLE_SINK_REPLICA;
+import static org.apache.seatunnel.connectors.seatunnel.assertion.sink.AssertSinkOptions.RULES;
 
 @AutoService(Factory.class)
 public class AssertSinkFactory implements TableSinkFactory {
@@ -38,10 +38,7 @@ public String factoryIdentifier() {
 
     @Override
     public OptionRule optionRule() {
-        return OptionRule.builder()
-                .required(RULES)
-                .optional(SinkCommonOptions.MULTI_TABLE_SINK_REPLICA)
-                .build();
+        return OptionRule.builder().required(RULES).optional(MULTI_TABLE_SINK_REPLICA).build();
     }
 
     @Override

seatunnel-connectors-v2/connector-assert/src/main/java/org/apache/seatunnel/connectors/seatunnel/assertion/sink/AssertSinkOptions.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.connectors.seatunnel.assertion.sink;
+
+import org.apache.seatunnel.shade.com.fasterxml.jackson.core.type.TypeReference;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.sink.SinkCommonOptions;
+
+import java.util.Map;
+
+public class AssertSinkOptions {
+
+    public static final Option<Map<String, Object>> RULES =
+            Options.key("rules")
+                    .type(new TypeReference<Map<String, Object>>() {})
+                    .noDefaultValue()
+                    .withDescription(
+                            "Rule definition of user's available data. Each rule represents one field validation or row num validation.");
+
+    public static Option<Integer> MULTI_TABLE_SINK_REPLICA =
+            SinkCommonOptions.MULTI_TABLE_SINK_REPLICA;
+}