[Feature][Connector-redis] fix redis cluster bug and add cluster e2e

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

@@ -32,6 +32,7 @@ Used to write data to Redis.
 | value_field        | string  | no                    | -             |
 | hash_key_field     | string  | no                    | -             |
 | hash_value_field   | string  | no                    | -             |
+| field_delimiter    | string  | no                    | ','           |
 | common-options     |         | no                    | -             |
 
 ### host [string]
@@ -119,7 +120,7 @@ redis nodes information, used in cluster mode, must like as the following format
 
 ### format [string]
 
-The format of upstream data, now only support `json`, `text` will be supported later, default `json`.
+The format of upstream data, currently support `json`, `text` format, default `json`.
 
 When you assign format is `json`, for example:
 
@@ -134,9 +135,18 @@ Connector will generate data as the following and write it to redis:
 ```json
 
 {"code":  200, "data":  "get success", "success":  "true"}
+```
 
+when you assign format is `text`, and set field_delimiter to `#`, connector will generate data as the following and write it to redis:
+```text
+200#get success#true
 ```
 
+### field_delimiter [string]
+Field delimiter, used to tell connector how to slice and dice fields.
+
+Currently, only need to be configured when format is `text`. default is ",".
+
 ### expire [long]
 
 Set redis expiration time, the unit is second. The default value is -1, keys do not automatically expire by default.
@@ -219,7 +229,7 @@ custom key:
 Redis {
   host = localhost
   port = 6379
-  key = "name:{name}"
+  key = "name:${name}"
   support_custom_key = true
   data_type = key
 }

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

@@ -37,6 +37,7 @@ Used to read data from Redis.
 | schema              | config | yes when format=json           | -             |
 | format              | string | no                             | json          |
 | single_field_name   | string | yes when read_key_enabled=true | -             |
+| field_delimiter     | string | no                             | ','           |
 | common-options      |        | no                             | -             |
 
 ### host [string]
@@ -252,21 +253,44 @@ connector will generate data as the following:
 | ---- | ----------- | ------- |
 | 200  | get success | true    |
 
-when you assign format is `text`, connector will do nothing for upstream data, for example:
+when you assign format is `text`, you can choose to specify the schema information or not. 
 
-upstream data is the following:
-
-```json
-{"code":  200, "data":  "get success", "success":  true}
+For example, upstream data is the following:
 
+```text
+200#get success#true
 ```
 
+If you do not assign data schema connector will treat the upstream data as the following:
+
+| content                                                  |
+| -------------------------------------------------------- |
+| 200#get success#true |
+
+If you assign data schema, you should also assign the option `schema` and `field_delimiter` as following:
+
+```hocon
+field_delimiter = "#"
+schema {
+    fields {
+        code = int
+        data = string
+        success = boolean
+    }
+}
+
+```
 connector will generate data as the following:
 
 | content                                                  |
 | -------------------------------------------------------- |
 | {"code":  200, "data":  "get success", "success":  true} |
 
+### field_delimiter [string]
+Field delimiter, used to tell connector how to slice and dice fields.
+
+Currently, only need to be configured when format is text. default is ",".
+
 ### schema [config]
 
 #### fields [config]

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

@@ -17,22 +17,23 @@ import ChangeLog from '../changelog/connector-redis.md';
 | name               | type    |       required        | default value |
 |--------------------|---------|-----------------------|---------------|
 | host               | string  | `mode=single`时必须      | -             |
-| port               | int     | no                    | 6379          |
-| key                | string  | yes                   | -             |
-| data_type          | string  | yes                   | -             |
-| batch_size         | int     | no                    | 10            |
-| user               | string  | no                    | -             |
-| auth               | string  | no                    | -             |
-| db_num             | int     | no                    | 0             |
-| mode               | string  | no                    | single        |
-| nodes              | list    | yes when mode=cluster | -             |
-| format             | string  | no                    | json          |
-| expire             | long    | no                    | -1            |
-| support_custom_key | boolean | no                    | false         |
-| value_field        | string  | no                    | -             |
-| hash_key_field     | string  | no                    | -             |
-| hash_value_field   | string  | no                    | -             |
-| common-options     |         | no                    | -             |
+| port               | int     | 否                 | 6379          |
+| key                | string  | 是                 | -             |
+| data_type          | string  | 是                 | -             |
+| batch_size         | int     | 否                 | 10            |
+| user               | string  | 否                 | -             |
+| auth               | string  | 否                 | -             |
+| db_num             | int     | 否                 | 0             |
+| mode               | string  | 否                 | single        |
+| nodes              | list    | `mode=cluster`时必须 | -             |
+| format             | string  | 否                 | json          |
+| expire             | long    | 否                 | -1            |
+| support_custom_key | boolean | 否                 | false         |
+| value_field        | string  | 否                 | -             |
+| hash_key_field     | string  | 否                 | -             |
+| hash_value_field   | string  | 否                 | -             |
+| field_delimiter    | string  | 否                 | ","           |
+| common-options     |         | 否                 | -             |
 
 ### host [string]
 
@@ -114,7 +115,7 @@ Redis 节点信息，在集群模式下使用，必须按如下格式：
 
 ### format [string]
 
-上游数据的格式，目前只支持 `json`，以后会支持 `text`，默认 `json`。
+上游数据的格式，目前只支持 `json`，`text`，默认 `json`。
 
 当你指定格式为 `json` 时，例如：
 
@@ -130,6 +131,18 @@ Redis 节点信息，在集群模式下使用，必须按如下格式：
 {"code":  200, "data":  "获取成功", "success":  "true"}
 ```
 
+当你指定format为`text`，并设置field_delimiter为`#`时，连接器将生成如下数据并将其写入redis：
+
+```text
+200#get success#true
+```
+
+### field_delimiter [string]
+字段分隔符，用于告诉连接器如何分割字段。
+
+目前仅当格式为text时需要配置。默认为","。
+
+
 ### expire [long]
 
 设置 Redis 的过期时间，单位为秒。默认值为 -1，表示键不会自动过期。
@@ -210,7 +223,7 @@ Redis {
 Redis {
   host = localhost
   port = 6379
-  key = "name:{name}"
+  key = "name:${name}"
   support_custom_key = true
   data_type = key
 }

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

@@ -34,6 +34,7 @@ import ChangeLog from '../changelog/connector-redis.md';
 | nodes               | list   | `mode=cluster` 时必须 | -      |
 | schema              | config | `format=json` 时必须  | -      |
 | format              | string | 否                  | json   |
+| field_delimiter     | string | 否                  | ','    |
 | common-options      |        | 否                  | -      |
 
 ### host [string]
@@ -203,21 +204,44 @@ schema {
 | ---- | ----------- | ------- |
 | 200  | get success | true    |
 
-当指定格式为 `text` 时，连接器不会对上游数据做任何处理，例如：
+当指定格式为 `text` 时，可以选择是否指定schema参数。
 
-当上游数据如下时：
+例如, 当上游数据如下时：
 
-```json
-{"code":  200, "data":  "get success", "success":  true}
+```text
+200#get success#true
+```
+
+如果不指定schema参数，连接器将按照以下方式处理上游数据：
+
+| content                                                  |
+| -------------------------------------------------------- |
+| 200#get success#true |
+
+如果指定schema参数，此时需要同时配置`schema`和`field_delimiter`，如下所示：
+```hocon
+field_delimiter = "#"
+schema {
+    fields {
+        code = int
+        data = string
+        success = boolean
+    }
+}
 
 ```
 
-连接器将会生成如下格式数据：
+连接器将生成如下数据：
 
 | content                                                  |
 | -------------------------------------------------------- |
 | {"code":  200, "data":  "get success", "success":  true} |
 
+### field_delimiter [string]
+字段分隔符，用于告诉连接器如何分割字段。
+
+目前仅当格式为text时需要配置。默认为","。
+
 ### schema [config]
 
 #### fields [config]

seatunnel-connectors-v2/connector-redis/pom.xml

@@ -47,6 +47,12 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-text</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>redis.clients</groupId>
             <artifactId>jedis</artifactId>

seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/client/RedisClient.java

@@ -30,7 +30,7 @@
 import java.util.Map;
 import java.util.Set;
 
-public abstract class RedisClient extends Jedis {
+public abstract class RedisClient {
 
     protected final RedisParameters redisParameters;
 
@@ -62,14 +62,14 @@ private ScanResult<String> scanByRedisVersion(
         if (redisVersion <= REDIS_5) {
             return scanOnRedis5(cursor, scanParams, type);
         } else {
-            return jedis.scan(cursor, scanParams, type.name());
+            return scanKeyResult(cursor, scanParams, type);
         }
     }
 
     // When the version is earlier than redis5, scan command does not support type
     private ScanResult<String> scanOnRedis5(
             String cursor, ScanParams scanParams, RedisDataType type) {
-        ScanResult<String> scanResult = jedis.scan(cursor, scanParams);
+        ScanResult<String> scanResult = scanKeyResult(cursor, scanParams, null);
         String resultCursor = scanResult.getCursor();
         List<String> keys = scanResult.getResult();
         List<String> typeKeys = new ArrayList<>(keys.size());
@@ -82,6 +82,15 @@ private ScanResult<String> scanOnRedis5(
         return new ScanResult<>(resultCursor, typeKeys);
     }
 
+    public void close() {
+        if (jedis != null) {
+            jedis.close();
+        }
+    }
+
+    public abstract ScanResult<String> scanKeyResult(
+            String cursor, ScanParams scanParams, RedisDataType type);
+
     public abstract List<String> batchGetString(List<String> keys);
 
     public abstract List<List<String>> batchGetList(List<String> keys);