[Feature][Connector-V2] Support TableSourceFactory/TableSinkFactory on redis

seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/sink/HttpSink.java

@@ -75,7 +75,7 @@ public HttpSink(Config pluginConfig, SeaTunnelRowType rowType) {
 
     @Override
     public String getPluginName() {
-        return "Http";
+        return HttpConfig.CONNECTOR_IDENTITY;
     }
 
     @Override

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisConfig.java

@@ -20,8 +20,12 @@
 import org.apache.seatunnel.api.configuration.Option;
 import org.apache.seatunnel.api.configuration.Options;
 
+import java.util.List;
+
 public class RedisConfig {
 
+    public static final String CONNECTOR_IDENTITY = "Redis";
+
     public enum RedisMode {
         SINGLE,
         CLUSTER;
@@ -38,8 +42,8 @@ public enum HashKeyParseMode {
                     .noDefaultValue()
                     .withDescription("redis hostname or ip");
 
-    public static final Option<String> PORT =
-            Options.key("port").stringType().noDefaultValue().withDescription("redis port");
+    public static final Option<Integer> PORT =
+            Options.key("port").intType().noDefaultValue().withDescription("redis port");
 
     public static final Option<String> AUTH =
             Options.key("auth")
@@ -95,9 +99,9 @@ public enum HashKeyParseMode {
                     .withDescription(
                             "redis mode, support single or cluster, default value is single");
 
-    public static final Option<String> NODES =
+    public static final Option<List<String>> NODES =
             Options.key("nodes")
-                    .stringType()
+                    .listType()
                     .noDefaultValue()
                     .withDescription(
                             "redis nodes information, used in cluster mode, must like as the following format: [host1:port1, host2:port2]");

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java

@@ -17,8 +17,7 @@
 
 package org.apache.seatunnel.connectors.seatunnel.redis.config;
 
-import org.apache.seatunnel.shade.com.typesafe.config.Config;
-
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
 import org.apache.seatunnel.connectors.seatunnel.redis.exception.RedisConnectorException;
 
@@ -50,57 +49,57 @@ public class RedisParameters implements Serializable {
     private List<String> redisNodes = Collections.emptyList();
     private long expire = RedisConfig.EXPIRE.defaultValue();
 
-    public void buildWithConfig(Config config) {
+    public void buildWithConfig(ReadonlyConfig config) {
         // set host
-        this.host = config.getString(RedisConfig.HOST.key());
+        this.host = config.get(RedisConfig.HOST);
         // set port
-        this.port = config.getInt(RedisConfig.PORT.key());
+        this.port = config.get(RedisConfig.PORT);
         // set auth
-        if (config.hasPath(RedisConfig.AUTH.key())) {
-            this.auth = config.getString(RedisConfig.AUTH.key());
+        if (config.getOptional(RedisConfig.AUTH).isPresent()) {
+            this.auth = config.get(RedisConfig.AUTH);
         }
         // set db_num
-        if (config.hasPath(RedisConfig.DB_NUM.key())) {
-            this.dbNum = config.getInt(RedisConfig.DB_NUM.key());
+        if (config.getOptional(RedisConfig.DB_NUM).isPresent()) {
+            this.dbNum = config.get(RedisConfig.DB_NUM);
         }
         // set user
-        if (config.hasPath(RedisConfig.USER.key())) {
-            this.user = config.getString(RedisConfig.USER.key());
+        if (config.getOptional(RedisConfig.USER).isPresent()) {
+            this.user = config.get(RedisConfig.USER);
         }
         // set mode
-        if (config.hasPath(RedisConfig.MODE.key())) {
+        if (config.getOptional(RedisConfig.MODE).isPresent()) {
             this.mode =
                     RedisConfig.RedisMode.valueOf(
-                            config.getString(RedisConfig.MODE.key()).toUpperCase());
+                            config.get(RedisConfig.MODE).name().toUpperCase());
         } else {
             this.mode = RedisConfig.MODE.defaultValue();
         }
         // set hash key mode
-        if (config.hasPath(RedisConfig.HASH_KEY_PARSE_MODE.key())) {
+        if (config.getOptional(RedisConfig.HASH_KEY_PARSE_MODE).isPresent()) {
             this.hashKeyParseMode =
                     RedisConfig.HashKeyParseMode.valueOf(
-                            config.getString(RedisConfig.HASH_KEY_PARSE_MODE.key()).toUpperCase());
+                            config.get(RedisConfig.HASH_KEY_PARSE_MODE).name().toUpperCase());
         } else {
             this.hashKeyParseMode = RedisConfig.HASH_KEY_PARSE_MODE.defaultValue();
         }
         // set redis nodes information
-        if (config.hasPath(RedisConfig.NODES.key())) {
-            this.redisNodes = config.getStringList(RedisConfig.NODES.key());
+        if (config.getOptional(RedisConfig.NODES).isPresent()) {
+            this.redisNodes = config.get(RedisConfig.NODES);
         }
         // set key
-        if (config.hasPath(RedisConfig.KEY.key())) {
-            this.keyField = config.getString(RedisConfig.KEY.key());
+        if (config.getOptional(RedisConfig.KEY).isPresent()) {
+            this.keyField = config.get(RedisConfig.KEY);
         }
         // set keysPattern
-        if (config.hasPath(RedisConfig.KEY_PATTERN.key())) {
-            this.keysPattern = config.getString(RedisConfig.KEY_PATTERN.key());
+        if (config.getOptional(RedisConfig.KEY_PATTERN).isPresent()) {
+            this.keysPattern = config.get(RedisConfig.KEY_PATTERN);
         }
-        if (config.hasPath(RedisConfig.EXPIRE.key())) {
-            this.expire = config.getLong(RedisConfig.EXPIRE.key());
+        if (config.getOptional(RedisConfig.EXPIRE).isPresent()) {
+            this.expire = config.get(RedisConfig.EXPIRE);
         }
         // set redis data type
         try {
-            String dataType = config.getString(RedisConfig.DATA_TYPE.key());
+            String dataType = config.get(RedisConfig.DATA_TYPE);
             this.redisDataType = RedisDataType.valueOf(dataType.toUpperCase());
         } catch (IllegalArgumentException e) {
             throw new RedisConnectorException(

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java

@@ -17,61 +17,34 @@
 
 package org.apache.seatunnel.connectors.seatunnel.redis.sink;
 
-import org.apache.seatunnel.shade.com.typesafe.config.Config;
-
-import org.apache.seatunnel.api.common.PrepareFailException;
-import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
-import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
-import org.apache.seatunnel.common.config.CheckConfigUtil;
-import org.apache.seatunnel.common.config.CheckResult;
-import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
 import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
 import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisConfig;
 import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisParameters;
-import org.apache.seatunnel.connectors.seatunnel.redis.exception.RedisConnectorException;
-
-import com.google.auto.service.AutoService;
 
 import java.io.IOException;
 
-@AutoService(SeaTunnelSink.class)
 public class RedisSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
     private final RedisParameters redisParameters = new RedisParameters();
     private SeaTunnelRowType seaTunnelRowType;
-    private Config pluginConfig;
-
-    @Override
-    public String getPluginName() {
-        return "Redis";
+    private ReadonlyConfig readonlyConfig;
+    private CatalogTable catalogTable;
+
+    public RedisSink(ReadonlyConfig config, CatalogTable table) {
+        this.readonlyConfig = config;
+        this.catalogTable = table;
+        this.redisParameters.buildWithConfig(config);
+        this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
     }
 
     @Override
-    public void prepare(Config pluginConfig) throws PrepareFailException {
-        this.pluginConfig = pluginConfig;
-        CheckResult result =
-                CheckConfigUtil.checkAllExists(
-                        pluginConfig,
-                        RedisConfig.HOST.key(),
-                        RedisConfig.PORT.key(),
-                        RedisConfig.KEY.key(),
-                        RedisConfig.DATA_TYPE.key());
-        if (!result.isSuccess()) {
-            throw new RedisConnectorException(
-                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
-                    String.format(
-                            "PluginName: %s, PluginType: %s, Message: %s",
-                            getPluginName(), PluginType.SINK, result.getMsg()));
-        }
-        this.redisParameters.buildWithConfig(pluginConfig);
-    }
-
-    @Override
-    public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
-        this.seaTunnelRowType = seaTunnelRowType;
+    public String getPluginName() {
+        return RedisConfig.CONNECTOR_IDENTITY;
     }
 
     @Override

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkFactory.java

@@ -18,8 +18,11 @@
 package org.apache.seatunnel.connectors.seatunnel.redis.sink;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+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 org.apache.seatunnel.connectors.seatunnel.redis.config.RedisConfig;
 
 import com.google.auto.service.AutoService;
@@ -31,6 +34,12 @@ public String factoryIdentifier() {
         return "Redis";
     }
 
+    @Override
+    public TableSink createSink(TableSinkFactoryContext context) {
+        CatalogTable catalogTable = context.getCatalogTable();
+        return () -> new RedisSink(context.getOptions(), catalogTable);
+    }
+
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSource.java

@@ -17,20 +17,15 @@
 
 package org.apache.seatunnel.connectors.seatunnel.redis.source;
 
-import org.apache.seatunnel.shade.com.typesafe.config.Config;
-
-import org.apache.seatunnel.api.common.PrepareFailException;
 import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.serialization.DeserializationSchema;
 import org.apache.seatunnel.api.source.Boundedness;
-import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
 import org.apache.seatunnel.api.table.catalog.schema.TableSchemaOptions;
-import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
-import org.apache.seatunnel.common.config.CheckConfigUtil;
-import org.apache.seatunnel.common.config.CheckResult;
 import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
 import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitSource;
@@ -40,40 +35,29 @@
 import org.apache.seatunnel.connectors.seatunnel.redis.exception.RedisConnectorException;
 import org.apache.seatunnel.format.json.JsonDeserializationSchema;
 
-import com.google.auto.service.AutoService;
+import com.google.common.collect.Lists;
+
+import java.util.List;
 
-@AutoService(SeaTunnelSource.class)
 public class RedisSource extends AbstractSingleSplitSource<SeaTunnelRow> {
     private final RedisParameters redisParameters = new RedisParameters();
     private SeaTunnelRowType seaTunnelRowType;
     private DeserializationSchema<SeaTunnelRow> deserializationSchema;
 
+    private CatalogTable catalogTable;
+
     @Override
     public String getPluginName() {
-        return "Redis";
+        return RedisConfig.CONNECTOR_IDENTITY;
     }
 
-    @Override
-    public void prepare(Config pluginConfig) throws PrepareFailException {
-        CheckResult result =
-                CheckConfigUtil.checkAllExists(
-                        pluginConfig,
-                        RedisConfig.HOST.key(),
-                        RedisConfig.PORT.key(),
-                        RedisConfig.KEY_PATTERN.key(),
-                        RedisConfig.DATA_TYPE.key());
-        if (!result.isSuccess()) {
-            throw new RedisConnectorException(
-                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
-                    String.format(
-                            "PluginName: %s, PluginType: %s, Message: %s",
-                            getPluginName(), PluginType.SOURCE, result.getMsg()));
-        }
-        this.redisParameters.buildWithConfig(pluginConfig);
+    public RedisSource(ReadonlyConfig readonlyConfig) {
+
+        this.redisParameters.buildWithConfig(readonlyConfig);
         // TODO: use format SPI
         // default use json format
-        if (pluginConfig.hasPath(RedisConfig.FORMAT.key())) {
-            if (!pluginConfig.hasPath(TableSchemaOptions.SCHEMA.key())) {
+        if (readonlyConfig.getOptional(RedisConfig.FORMAT).isPresent()) {
+            if (!readonlyConfig.getOptional(TableSchemaOptions.SCHEMA).isPresent()) {
                 throw new RedisConnectorException(
                         SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
                         String.format(
@@ -85,15 +69,16 @@ public void prepare(Config pluginConfig) throws PrepareFailException {
 
             RedisConfig.Format format =
                     RedisConfig.Format.valueOf(
-                            pluginConfig.getString(RedisConfig.FORMAT.key()).toUpperCase());
+                            readonlyConfig.get(RedisConfig.FORMAT).name().toUpperCase());
             if (RedisConfig.Format.JSON.equals(format)) {
-                this.seaTunnelRowType =
-                        CatalogTableUtil.buildWithConfig(pluginConfig).getSeaTunnelRowType();
+                this.catalogTable = CatalogTableUtil.buildWithConfig(readonlyConfig);
+                this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
                 this.deserializationSchema =
                         new JsonDeserializationSchema(false, false, seaTunnelRowType);
             }
         } else {
-            this.seaTunnelRowType = CatalogTableUtil.buildSimpleTextSchema();
+            this.catalogTable = CatalogTableUtil.buildSimpleTextTable();
+            this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
             this.deserializationSchema = null;
         }
     }
@@ -104,8 +89,8 @@ public Boundedness getBoundedness() {
     }
 
     @Override
-    public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
-        return seaTunnelRowType;
+    public List<CatalogTable> getProducedCatalogTables() {
+        return Lists.newArrayList(catalogTable);
     }
 
     @Override

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java

@@ -19,31 +19,45 @@
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.source.SourceSplit;
 import org.apache.seatunnel.api.table.catalog.schema.TableSchemaOptions;
+import org.apache.seatunnel.api.table.connector.TableSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
 import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisConfig;
 
 import com.google.auto.service.AutoService;
 
+import java.io.Serializable;
+
 @AutoService(Factory.class)
 public class RedisSourceFactory implements TableSourceFactory {
     @Override
     public String factoryIdentifier() {
         return "Redis";
     }
 
+    @Override
+    public <T, SplitT extends SourceSplit, StateT extends Serializable>
+            TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
+        return () -> (SeaTunnelSource<T, SplitT, StateT>) new RedisSource(context.getOptions());
+    }
+
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
                 .required(
-                        RedisConfig.HOST, RedisConfig.PORT, RedisConfig.KEY, RedisConfig.DATA_TYPE)
+                        RedisConfig.HOST,
+                        RedisConfig.PORT,
+                        RedisConfig.KEY_PATTERN,
+                        RedisConfig.DATA_TYPE)
                 .optional(
                         RedisConfig.MODE,
                         RedisConfig.HASH_KEY_PARSE_MODE,
                         RedisConfig.AUTH,
                         RedisConfig.USER,
-                        RedisConfig.KEY_PATTERN)
+                        RedisConfig.KEY)
                 .conditional(RedisConfig.MODE, RedisConfig.RedisMode.CLUSTER, RedisConfig.NODES)
                 .bundled(RedisConfig.FORMAT, TableSchemaOptions.SCHEMA)
                 .build();