[Improve][Connector-v2] Redis support select db

docs/en/connector-v2/sink/Redis.md

@@ -20,6 +20,7 @@ Used to write data to Redis.
 | data_type      | string | yes                   | -             |
 | user           | string | no                    | -             |
 | auth           | string | no                    | -             |
+| db_num         | int    | no                    | -             |
 | mode           | string | no                    | single        |
 | nodes          | list   | yes when mode=cluster | -             |
 | format         | string | no                    | json          |
@@ -91,6 +92,10 @@ redis authentication user, you need it when you connect to an encrypted cluster
 
 Redis authentication password, you need it when you connect to an encrypted cluster
 
+### db_num [int]
+
+Redis database index ID. It is connected to db 0 by default
+
 ### mode [string]
 
 redis mode, `single` or `cluster`, default is `single`

docs/en/connector-v2/source/Redis.md

@@ -25,6 +25,7 @@ Used to read data from Redis.
 | data_type           | string | yes                   | -             |
 | user                | string | no                    | -             |
 | auth                | string | no                    | -             |
+| db_num              | int    | no                    | -             |
 | mode                | string | no                    | single        |
 | hash_key_parse_mode | string | no                    | all           |
 | nodes               | list   | yes when mode=cluster | -             |
@@ -151,6 +152,10 @@ redis authentication user, you need it when you connect to an encrypted cluster
 
 redis authentication password, you need it when you connect to an encrypted cluster
 
+### db_num [int]
+
+Redis database index ID. It is connected to db 0 by default
+
 ### mode [string]
 
 redis mode, `single` or `cluster`, default is `single`

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

@@ -48,6 +48,13 @@ public enum HashKeyParseMode {
                     .withDescription(
                             "redis authentication password, you need it when you connect to an encrypted cluster");
 
+    public static final Option<Integer> DB_NUM =
+            Options.key("db_num")
+                    .intType()
+                    .defaultValue(0)
+                    .withDescription(
+                            "redis  database index id, It is connected to db 0 by default");
+
     public static final Option<String> USER =
             Options.key("user")
                     .stringType()

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

@@ -40,6 +40,7 @@ public class RedisParameters implements Serializable {
     private String host;
     private int port;
     private String auth = "";
+    private Integer dbNum;
     private String user = "";
     private String keysPattern;
     private String keyField;
@@ -58,6 +59,10 @@ public void buildWithConfig(Config config) {
         if (config.hasPath(RedisConfig.AUTH.key())) {
             this.auth = config.getString(RedisConfig.AUTH.key());
         }
+        // set db_num
+        if (config.hasPath(RedisConfig.DB_NUM.key())) {
+            this.dbNum = config.getInt(RedisConfig.DB_NUM.key());
+        }
         // set user
         if (config.hasPath(RedisConfig.USER.key())) {
             this.user = config.getString(RedisConfig.USER.key());
@@ -115,6 +120,9 @@ public Jedis buildJedis() {
                 if (StringUtils.isNotBlank(user)) {
                     jedis.aclSetUser(user);
                 }
+                if (dbNum != null && dbNum > 0) {
+                    jedis.select(dbNum);
+                }
                 return jedis;
             case CLUSTER:
                 HashSet<HostAndPort> nodes = new HashSet<>();
@@ -148,7 +156,11 @@ public Jedis buildJedis() {
                 } else {
                     jedisCluster = new JedisCluster(nodes);
                 }
-                return new JedisWrapper(jedisCluster);
+                JedisWrapper jedisWrapper = new JedisWrapper(jedisCluster);
+                if (dbNum != null && dbNum > 0) {
+                    jedisWrapper.select(dbNum);
+                }
+                return jedisWrapper;
             default:
                 // do nothing
                 throw new RedisConnectorException(