[Feature][scaleph-plugin-seatunnel-connectors] upgrade seatunnel connectors to 2.3.8

.github/workflows/release-manual-docker-seatunnel.yml

@@ -22,11 +22,11 @@ on:
       seatunnelVersion:
         description: 'seatunnel version'
         required: true
-        default: '2.3.7'
+        default: '2.3.8'
         type: choice
         options:
+          - 2.3.8
           - 2.3.7
-          - 2.3.6
       flinkVersion:
         description: 'flink version'
         required: true

README.md

@@ -13,7 +13,7 @@ Scaleph is driven by personal interest and evolves actively through faithful dev
 * Data Integration
   * Web-ui click-and-drag data integration ways backended by [Apache SeaTunnel](https://seatunnel.apache.org/) on Flink engine.
 
-  * Support the latest 2.3.7 V2 out-of-the-box connectors and transforms.
+  * Support the latest 2.3.8 V2 out-of-the-box connectors and transforms.
 
   * DataSource management.
 

scaleph-application/scaleph-application-flink/src/main/java/cn/sliew/scaleph/application/flink/resource/handler/FlinkImageMapping.java

@@ -38,7 +38,7 @@ public enum FlinkImageMapping {
     SQL_1_17(FlinkJobType.SQL, OperatorFlinkVersion.v1_17, FlinkVersionMapping.V_1_17, "ghcr.io/flowerfine/scaleph-sql-template:1.17"),
     SQL_1_18(FlinkJobType.SQL, OperatorFlinkVersion.v1_18, FlinkVersionMapping.V_1_18, "ghcr.io/flowerfine/scaleph-sql-template:1.18"),
 
-    SEATUNNEL_1_16(FlinkJobType.SEATUNNEL, OperatorFlinkVersion.v1_16, FlinkVersionMapping.V_1_16, "ghcr.io/flowerfine/scaleph-seatunnel:2.3.7-flink-1.16"),
+    SEATUNNEL_1_16(FlinkJobType.SEATUNNEL, OperatorFlinkVersion.v1_16, FlinkVersionMapping.V_1_16, "ghcr.io/flowerfine/scaleph-seatunnel:2.3.8-flink-1.16"),
     FLINK_CDC_1_18(FlinkJobType.FLINK_CDC, OperatorFlinkVersion.v1_18, FlinkVersionMapping.V_1_18, "ghcr.io/flowerfine/scaleph-flink-cdc:3.0.0-flink-1.18"),
     ;
 

scaleph-common/src/main/java/cn/sliew/scaleph/common/dict/seatunnel/SeaTunnelPluginMapping.java

@@ -46,6 +46,7 @@ public enum SeaTunnelPluginMapping {
     SOURCE_HTTP(SEATUNNEL, SOURCE, HTTP, "connector-http-base", BETA, BATCH, STREAM, SCHEMA_PROJECTION),
     SINK_HTTP(SEATUNNEL, SINK, HTTP, "connector-http-base", BETA),
     SINK_FEISHU(SEATUNNEL, SINK, FEISHU, "connector-http-feishu", ALPHA),
+    @Deprecated
     SINK_WECHAT(SEATUNNEL, SINK, WECHAT, "connector-http-wechat", ALPHA),
     SINK_DINGTALK(SEATUNNEL, SINK, DINGTALK, "connector-dingtalk", ALPHA),
     SOURCE_MYHOURS(SEATUNNEL, SOURCE, MYHOURS, "connector-http-myhours", ALPHA, BATCH, SCHEMA_PROJECTION),
@@ -86,6 +87,7 @@ public enum SeaTunnelPluginMapping {
     SINK_AMAZON_SQS(SEATUNNEL, SINK, AMAZON_SQS, "connector-amazonsqs", UNKNOWN),
     SOURCE_RABBITMQ(SEATUNNEL, SOURCE, RABBITMQ, "connector-rabbitmq", BETA, STREAM, EXACTLY_ONCE, SCHEMA_PROJECTION),
     SINK_RABBITMQ(SEATUNNEL, SINK, RABBITMQ, "connector-rabbitmq", BETA),
+    SINK_ACTIVEMQ(SEATUNNEL, SINK, ACTIVEMQ, "connector-activemq", BETA),
 
     SOURCE_JDBC(SEATUNNEL, SOURCE, JDBC, "connector-jdbc", GA, BATCH, SCHEMA_PROJECTION, PARALLELISM, SUPPORT_USER_DEFINED_SPLIT),
     SINK_JDBC(SEATUNNEL, SINK, JDBC, "connector-jdbc", GA, EXACTLY_ONCE, CDC),
@@ -101,13 +103,15 @@ public enum SeaTunnelPluginMapping {
     SINK_AMAZON_DYNAMODB(SEATUNNEL, SINK, AMAZON_DYNAMODB, "connector-amazondynamodb", BETA),
     SOURCE_CASSANDRA(SEATUNNEL, SOURCE, CASSANDRA, "connector-cassandra", BETA, BATCH, SCHEMA_PROJECTION),
     SINK_CASSANDRA(SEATUNNEL, SINK, CASSANDRA, "connector-cassandra", BETA),
+    SOURCE_TABLESTORE(SEATUNNEL, SOURCE, TABLESTORE, "connector-tablestore", ALPHA),
     SINK_TABLESTORE(SEATUNNEL, SINK, TABLESTORE, "connector-tablestore", ALPHA),
     SINK_GOOGLE_FIRE_STORE(SEATUNNEL, SINK, GOOGLE_FIRE_STORE, "connector-google-firestore", UNKNOWN),
 
     SOURCE_MYSQL_CDC(SEATUNNEL, SOURCE, MYSQL_CDC, "connector-cdc-mysql", GA, STREAM, EXACTLY_ONCE, PARALLELISM, SUPPORT_USER_DEFINED_SPLIT),
     SOURCE_SQLSERVER_CDC(SEATUNNEL, SOURCE, SQLSERVER_CDC, "connector-cdc-sqlserver", GA, STREAM, EXACTLY_ONCE, PARALLELISM, SUPPORT_USER_DEFINED_SPLIT),
     SOURCE_ORACLE_CDC(SEATUNNEL, SOURCE, ORACLE_CDC, "connector-cdc-oracle", UNKNOWN, STREAM, EXACTLY_ONCE, PARALLELISM, SUPPORT_USER_DEFINED_SPLIT),
     SOURCE_POSTGRESQL_CDC(SEATUNNEL, SOURCE, POSTGRESQL_CDC, "connector-cdc-postgres", UNKNOWN, STREAM, EXACTLY_ONCE, PARALLELISM, SUPPORT_USER_DEFINED_SPLIT),
+    SOURCE_OPENGAUSS_CDC(SEATUNNEL, SOURCE, OPENGAUSS_CDC, "connector-cdc-opengauss", UNKNOWN, STREAM, EXACTLY_ONCE, PARALLELISM, SUPPORT_USER_DEFINED_SPLIT),
     SOURCE_MONGODB_CDC(SEATUNNEL, SOURCE, MONGODB_CDC, "connector-cdc-mongodb", UNKNOWN, STREAM, EXACTLY_ONCE, PARALLELISM, SUPPORT_USER_DEFINED_SPLIT),
 
     SOURCE_HIVE(SEATUNNEL, SOURCE, HIVE, "connector-hive", GA, BATCH, EXACTLY_ONCE, SCHEMA_PROJECTION, PARALLELISM),
@@ -128,9 +132,11 @@ public enum SeaTunnelPluginMapping {
     SINK_S3REDSHIFT(SEATUNNEL, SINK, S3REDSHIFT, "connector-s3-redshift", GA, EXACTLY_ONCE),
     SOURCE_MAXCOMPUTE(SEATUNNEL, SOURCE, MAXCOMPUTE, "connector-maxcompute", ALPHA, BATCH, PARALLELISM),
     SINK_MAXCOMPUTE(SEATUNNEL, SINK, MAXCOMPUTE, "connector-maxcompute", ALPHA),
+    SOURCE_HBASE(SEATUNNEL, SOURCE, HBASE, "connector-hbase", ALPHA),
     SINK_HBASE(SEATUNNEL, SINK, HBASE, "connector-hbase", ALPHA),
     SOURCE_KUDU(SEATUNNEL, SOURCE, KUDU, "connector-kudu", BETA, BATCH),
     SINK_KUDU(SEATUNNEL, SINK, KUDU, "connector-kudu", BETA),
+    SINK_DRUID(SEATUNNEL, SINK, DRUID, "connector-druid", BETA),
     SOURCE_IOTDB(SEATUNNEL, SOURCE, IOTDB, "connector-iotdb", GA, BATCH, EXACTLY_ONCE, SCHEMA_PROJECTION, PARALLELISM),
     SINK_IOTDB(SEATUNNEL, SINK, IOTDB, "connector-iotdb", GA, EXACTLY_ONCE),
     SOURCE_OPENMLDB(SEATUNNEL, SOURCE, OPENMLDB, "connector-openmldb", BETA, BATCH, STREAM),
@@ -140,17 +146,25 @@ public enum SeaTunnelPluginMapping {
     SINK_INFLUXDB(SEATUNNEL, SINK, INFLUXDB, "connector-influxdb", BETA),
     SOURCE_TDENGINE(SEATUNNEL, SOURCE, TDENGINE, "connector-tdengine", BETA),
     SINK_TDENGINE(SEATUNNEL, SINK, TDENGINE, "connector-tdengine", BETA),
+    SOURCE_SLS(SEATUNNEL, SOURCE, SLS, "connector-sls", BETA),
+
+    SOURCE_MILVUS(SEATUNNEL, SOURCE, MILVUS, "connector-milvus", UNKNOWN),
+    SINK_MILVUS(SEATUNNEL, SINK, MILVUS, "connector-milvus", UNKNOWN),
+    SOURCE_WEB3J(SEATUNNEL, SOURCE, WEB3J, "connector-web3j", UNKNOWN),
 
     SINK_SENTRY(SEATUNNEL, SINK, SENTRY, "connector-sentry", ALPHA),
     SOURCE_GOOGLE_SHEETS(SEATUNNEL, SOURCE, GOOGLE_SHEETS, "connector-google-sheets", UNKNOWN, BATCH, SCHEMA_PROJECTION),
 
-    TRANSFORM_COPY(SEATUNNEL, TRANSFORM, COPY, "connector-copy", UNKNOWN),
-    TRANSFORM_FIELD_MAPPER(SEATUNNEL, TRANSFORM, FIELD_MAPPER, "connector-field-mapper", UNKNOWN),
-    TRANSFORM_FILTER_ROW_KIND(SEATUNNEL, TRANSFORM, FILTER_ROW_KIND, "connector-field-row-kind", UNKNOWN),
-    TRANSFORM_FILTER(SEATUNNEL, TRANSFORM, FILTER, "connector-filter", UNKNOWN),
-    TRANSFORM_REPLACE(SEATUNNEL, TRANSFORM, REPLACE, "connector-replace", UNKNOWN),
-    TRANSFORM_SPLIT(SEATUNNEL, TRANSFORM, SPLIT, "connector-split", UNKNOWN),
-    TRANSFORM_SQL(SEATUNNEL, TRANSFORM, SQL, "connector-sql", UNKNOWN),
+    TRANSFORM_COPY(SEATUNNEL, TRANSFORM, COPY, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_FIELD_MAPPER(SEATUNNEL, TRANSFORM, FIELD_MAPPER, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_FILTER_ROW_KIND(SEATUNNEL, TRANSFORM, FILTER_ROW_KIND, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_FILTER(SEATUNNEL, TRANSFORM, FILTER, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_REPLACE(SEATUNNEL, TRANSFORM, REPLACE, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_SPLIT(SEATUNNEL, TRANSFORM, SPLIT, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_SQL(SEATUNNEL, TRANSFORM, SQL, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_JSON_PATH(SEATUNNEL, TRANSFORM, JSON_PATH, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_DYNAMIC_COMPILE(SEATUNNEL, TRANSFORM, DYNAMIC_COMPILE, "seatunnel-transforms-v2", UNKNOWN),
+    TRANSFORM_LLM(SEATUNNEL, TRANSFORM, LLM, "seatunnel-transforms-v2", UNKNOWN),
     ;
 
     private SeaTunnelEngineType engineType;

scaleph-common/src/main/java/cn/sliew/scaleph/common/dict/seatunnel/SeaTunnelPluginName.java

@@ -36,7 +36,6 @@ public enum SeaTunnelPluginName implements DictInstance {
     SLACK("SlackSink", "Slack"),
     HTTP("Http", "Http"),
     FEISHU("Feishu", "Feishu"),
-
     @Deprecated
     WECHAT("WeChat", "WeChat"),
 
@@ -83,6 +82,7 @@ public enum SeaTunnelPluginName implements DictInstance {
     SQLSERVER_CDC("SqlServer-CDC", "SqlServer-CDC"),
     ORACLE_CDC("Oracle-CDC", "Oracle-CDC"),
     POSTGRESQL_CDC("Postgres-CDC", "PostgreSQL-CDC"),
+    OPENGAUSS_CDC("Opengauss-CDC", "Opengauss-CDC"),
     MONGODB_CDC("MongoDB-CDC", "MongoDB-CDC"),
 
     HIVE("Hive", "Hive"),

scaleph-common/src/main/java/cn/sliew/scaleph/common/dict/seatunnel/SeaTunnelVersion.java

@@ -33,6 +33,7 @@ public enum SeaTunnelVersion implements DictInstance {
     V_2_3_5("2.3.5", "2.3.5"),
     V_2_3_6("2.3.6", "2.3.6"),
     V_2_3_7("2.3.7", "2.3.7"),
+    V_2_3_8("2.3.8", "2.3.8"),
     ;
 
     @JsonCreator

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/cassandra/source/CassandraSourcePlugin.java

@@ -17,9 +17,10 @@
  */
 package cn.sliew.scaleph.plugin.seatunnel.flink.connectors.cassandra.source;
 
+import cn.sliew.carp.module.datasource.modal.DataSourceInfo;
+import cn.sliew.carp.module.datasource.modal.nosql.CassandraDataSourceProperties;
+import cn.sliew.milky.common.util.JacksonUtil;
 import cn.sliew.scaleph.common.dict.seatunnel.SeaTunnelPluginMapping;
-import cn.sliew.scaleph.ds.modal.AbstractDataSource;
-import cn.sliew.scaleph.ds.modal.nosql.CassandraDataSource;
 import cn.sliew.scaleph.plugin.framework.core.PluginInfo;
 import cn.sliew.scaleph.plugin.framework.property.PropertyDescriptor;
 import cn.sliew.scaleph.plugin.seatunnel.flink.SeaTunnelConnectorPlugin;
@@ -63,17 +64,19 @@ public List<ResourceProperty> getRequiredResources() {
     public ObjectNode createConf() {
         ObjectNode conf = super.createConf();
         JsonNode jsonNode = properties.get(ResourceProperties.DATASOURCE);
-        CassandraDataSource dataSource = (CassandraDataSource) AbstractDataSource.fromDsInfo((ObjectNode) jsonNode);
-        conf.putPOJO(HOST.getName(), dataSource.getHost());
-        conf.putPOJO(KEYSPACE.getName(), dataSource.getKeyspace());
-        if (StringUtils.hasText(dataSource.getUsername())) {
-            conf.putPOJO(USERNAME.getName(), dataSource.getUsername());
+
+        DataSourceInfo dataSourceInfo = JacksonUtil.toObject(jsonNode, DataSourceInfo.class);
+        CassandraDataSourceProperties props = (CassandraDataSourceProperties) dataSourceInfo.getProps();
+        conf.putPOJO(HOST.getName(), props.getHost());
+        conf.putPOJO(KEYSPACE.getName(), props.getKeyspace());
+        if (StringUtils.hasText(props.getUsername())) {
+            conf.putPOJO(USERNAME.getName(), props.getUsername());
         }
-        if (StringUtils.hasText(dataSource.getPassword())) {
-            conf.putPOJO(PASSWORD.getName(), dataSource.getPassword());
+        if (StringUtils.hasText(props.getPassword())) {
+            conf.putPOJO(PASSWORD.getName(), props.getPassword());
         }
-        if (StringUtils.hasText(dataSource.getDatacenter())) {
-            conf.putPOJO(DATACENTER.getName(), dataSource.getDatacenter());
+        if (StringUtils.hasText(props.getDatacenter())) {
+            conf.putPOJO(DATACENTER.getName(), props.getDatacenter());
         }
         return conf;
     }

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/cdc/oracle/source/OracleCDCSourcePlugin.java

@@ -45,6 +45,8 @@ public OracleCDCSourcePlugin() {
         props.add(PASSWORD);
         props.add(DATABASE);
         props.add(OracleCDCSourceProperties.SCHEMA);
+        props.add(OracleCDCSourceProperties.USE_SELECT_COUNT);
+        props.add(OracleCDCSourceProperties.SKIP_ANALYZE);
         props.add(TABLE);
         props.add(TABLE_CONFIG);
         props.add(STARTUP_MODE);

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/cdc/oracle/source/OracleCDCSourceProperties.java

@@ -35,4 +35,20 @@ public enum OracleCDCSourceProperties {
             .parser(Parsers.STRING_ARRAY_PARSER)
             .addValidator(Validators.NON_BLANK_VALIDATOR)
             .validateAndBuild();
+
+    public static final PropertyDescriptor<Boolean> USE_SELECT_COUNT = new PropertyDescriptor.Builder()
+            .name("use_select_count")
+            .description("Use select count for table count rather then other methods in full stage.In this scenario, select count directly is used when it is faster to update statistics using sql from analysis table")
+            .type(PropertyType.BOOLEAN)
+            .parser(Parsers.BOOLEAN_PARSER)
+            .addValidator(Validators.BOOLEAN_VALIDATOR)
+            .validateAndBuild();
+
+    public static final PropertyDescriptor<Boolean> SKIP_ANALYZE = new PropertyDescriptor.Builder()
+            .name("skip_analyze")
+            .description("Skip the analysis of table count in full stage.In this scenario, you schedule analysis table sql to update related table statistics periodically or your table data does not change frequently")
+            .type(PropertyType.BOOLEAN)
+            .parser(Parsers.BOOLEAN_PARSER)
+            .addValidator(Validators.BOOLEAN_VALIDATOR)
+            .validateAndBuild();
 }

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/clickhosue/ClickHouseProperties.java

@@ -19,6 +19,7 @@
 package cn.sliew.scaleph.plugin.seatunnel.flink.connectors.clickhosue;
 
 import cn.sliew.scaleph.plugin.framework.property.*;
+import org.codehaus.jackson.JsonNode;
 
 public enum ClickHouseProperties {
     ;
@@ -60,4 +61,13 @@ public enum ClickHouseProperties {
             .addValidator(Validators.NON_BLANK_VALIDATOR)
             .validateAndBuild();
 
+    public static final PropertyDescriptor<JsonNode> CLICKHOUSE_CONFIG = new PropertyDescriptor.Builder()
+            .name("clickhouse.config")
+            .description(
+                    "clickhouse.* The way to specify the parameter is to add the prefix clickhouse. to the original parameter name. For example, the way to specify socket_timeout is: clickhouse.socket_timeout = 50000 . "
+                            + "If these non-essential parameters are not specified, they will use the default values given by clickhouse-jdbc.\n")
+            .type(PropertyType.OBJECT)
+            .parser(Parsers.JSON_PARSER)
+            .validateAndBuild();
+
 }

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/clickhosue/sink/ClickHouseSinkProperties.java

@@ -33,15 +33,6 @@ public enum ClickHouseSinkProperties {
             .addValidator(Validators.NON_BLANK_VALIDATOR)
             .validateAndBuild();
 
-    public static final PropertyDescriptor<JsonNode> CLICKHOUSE_CONFIG = new PropertyDescriptor.Builder()
-            .name("clickhouse.config")
-            .description(
-                    "clickhouse.* The way to specify the parameter is to add the prefix clickhouse. to the original parameter name. For example, the way to specify socket_timeout is: clickhouse.socket_timeout = 50000 . "
-                            + "If these non-essential parameters are not specified, they will use the default values given by clickhouse-jdbc.\n")
-            .type(PropertyType.OBJECT)
-            .parser(Parsers.JSON_PARSER)
-            .validateAndBuild();
-
     public static final PropertyDescriptor<Integer> BULK_SIZE = new PropertyDescriptor.Builder()
             .name("bulk_size")
             .description(

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/clickhosue/source/ClickHouseSourcePlugin.java

@@ -50,6 +50,7 @@ public ClickHouseSourcePlugin() {
         final List<PropertyDescriptor> props = new ArrayList<>();
         props.add(SQL);
         props.add(SERVER_TIME_ZONE);
+        props.add(CLICKHOUSE_CONFIG);
         props.add(CommonProperties.PARALLELISM);
         props.add(CommonProperties.RESULT_TABLE_NAME);
         supportedProperties = Collections.unmodifiableList(props);

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/doris/source/DorisSourcePlugin.java

@@ -37,6 +37,7 @@
 import java.util.List;
 
 import static cn.sliew.scaleph.plugin.seatunnel.flink.connectors.doris.DorisProperties.*;
+import static cn.sliew.scaleph.plugin.seatunnel.flink.connectors.doris.DorisProperties.DORIS_BATCH_SIZE;
 import static cn.sliew.scaleph.plugin.seatunnel.flink.connectors.doris.source.DorisSourceProperties.*;
 
 @AutoService(SeaTunnelConnectorPlugin.class)
@@ -51,6 +52,7 @@ public DorisSourcePlugin() {
         props.add(TABLE);
         props.add(DORIS_READ_FIELD);
         props.add(DORIS_FILTER_QUERY);
+        props.add(DORIS_BATCH_SIZE);
         props.add(DORIS_REQUEST_CONNECT_TIMEOUT_MS);
         props.add(DORIS_REQUEST_QUERY_TIMEOUT_S);
         props.add(DORIS_REQUEST_READ_TIMEOUT_MS);

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/elasticsearch/source/ElasticsearchSourcePlugin.java

@@ -47,8 +47,10 @@ public ElasticsearchSourcePlugin() {
                 "Used to read data from Elasticsearch. support version >= 2.x and < 8.x.",
                 ElasticsearchSourcePlugin.class.getName());
         final List<PropertyDescriptor> props = new ArrayList<>();
+        props.add(INDEX_LIST);
         props.add(INDEX);
         props.add(SOURCE);
+        props.add(ARRAY_COLUMN);
         props.add(QUERY);
         props.add(SCROLL_TIME);
         props.add(SCROLL_SIZE);

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/elasticsearch/source/ElasticsearchSourceProperties.java

@@ -27,6 +27,14 @@
 public enum ElasticsearchSourceProperties {
     ;
 
+    public static final PropertyDescriptor<JsonNode> INDEX_LIST = new PropertyDescriptor.Builder()
+            .name("index_list")
+            .description("The index_list is used to define multi-index synchronization tasks")
+            .type(PropertyType.OBJECT)
+            .parser(Parsers.JSON_PARSER)
+            .addValidator(Validators.NON_BLANK_VALIDATOR)
+            .validateAndBuild();
+
     public static final PropertyDescriptor<JsonNode> SOURCE = new PropertyDescriptor.Builder()
             .name("source")
             .description("The fields of index.")
@@ -35,6 +43,14 @@ public enum ElasticsearchSourceProperties {
             .addValidator(Validators.NON_BLANK_VALIDATOR)
             .validateAndBuild();
 
+    public static final PropertyDescriptor<String> ARRAY_COLUMN = new PropertyDescriptor.Builder()
+            .name("array_column")
+            .description("The fields of array type")
+            .type(PropertyType.STRING)
+            .parser(Parsers.STRING_PARSER)
+            .addValidator(Validators.NON_BLANK_VALIDATOR)
+            .validateAndBuild();
+
     public static final PropertyDescriptor<JsonNode> QUERY = new PropertyDescriptor.Builder()
             .name("query")
             .description("Elasticsearch DSL. You can control the range of data read.")

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/email/sink/EmailSinkPlugin.java

@@ -43,6 +43,7 @@ public EmailSinkPlugin() {
         props.add(EMAIL_TRANSPORT_PROTOCOL);
         props.add(EMAIL_FROM_ADDRESS);
         props.add(EMAIL_SMTP_AUTH);
+        props.add(EMAIL_SMTP_PORT);
         props.add(EMAIL_AUTHORIZATION_CODE);
         props.add(EMAIL_TO_ADDRESS);
         props.add(EMAIL_MESSAGE_HEADLINE);

scaleph-plugins/scaleph-plugin-seatunnel-connectors/src/main/java/cn/sliew/scaleph/plugin/seatunnel/flink/connectors/email/sink/EmailSinkProperties.java

@@ -19,6 +19,7 @@
 package cn.sliew.scaleph.plugin.seatunnel.flink.connectors.email.sink;
 
 import cn.sliew.scaleph.plugin.framework.property.*;
+import com.amazonaws.services.dynamodbv2.xspec.BOOL;
 
 public enum EmailSinkProperties {
     ;
@@ -50,13 +51,20 @@ public enum EmailSinkProperties {
             .addValidator(Validators.NON_BLANK_VALIDATOR)
             .validateAndBuild();
 
-    public static final PropertyDescriptor<String> EMAIL_SMTP_AUTH = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor<BOOL> EMAIL_SMTP_AUTH = new PropertyDescriptor.Builder()
             .name("email_smtp_auth")
             .description("Whether to authenticate the customer")
-            .type(PropertyType.STRING)
-            .parser(Parsers.STRING_PARSER)
-            .properties(Property.Required)
-            .addValidator(Validators.NON_BLANK_VALIDATOR)
+            .type(PropertyType.BOOLEAN)
+            .parser(Parsers.BOOLEAN_PARSER)
+            .addValidator(Validators.BOOLEAN_VALIDATOR)
+            .validateAndBuild();
+
+    public static final PropertyDescriptor<Integer> EMAIL_SMTP_PORT = new PropertyDescriptor.Builder()
+            .name("email_smtp_port")
+            .description("Select port for authentication.")
+            .type(PropertyType.INT)
+            .parser(Parsers.INTEGER_PARSER)
+            .addValidator(Validators.INTEGER_VALIDATOR)
             .validateAndBuild();
 
     public static final PropertyDescriptor<String> EMAIL_AUTHORIZATION_CODE = new PropertyDescriptor.Builder()