Add Option Config OUTPUT_KEY_NAME

Author: FrommyMind

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

@@ -83,7 +83,8 @@ public enum HashKeyParseMode {
             Options.key("output_key")
                     .stringType()
                     .noDefaultValue()
-                    .withDescription("The key column name. If not set will not output key as a new column.");
+                    .withDescription(
+                            "The key column name. If not set will not output key as a new column.");
 
     public static final Option<RedisDataType> DATA_TYPE =
             Options.key("data_type")

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java

@@ -229,6 +229,41 @@ public void testRedis(TestContainer container) throws IOException, InterruptedEx
         Assertions.assertEquals(0, jedis.llen("key_list"));
     }
 
+    @TestTemplate
+    public void testRedisWithOutputKey(TestContainer container)
+            throws IOException, InterruptedException {
+        Container.ExecResult execResult =
+                container.executeJob("/redis-with-output-key-to-redis.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(100, jedis.llen("key_list"));
+        List<String> keyList = jedis.lrange("key_list", 0, 100);
+        for (String key : keyList) {
+            //            each content should have a new column called "origin_key"
+            log.info(key);
+            Assertions.assertTrue(key.contains("origin_key"));
+        }
+        // Clear data to prevent data duplication in the next TestContainer
+        jedis.del("key_list");
+        Assertions.assertEquals(0, jedis.llen("key_list"));
+    }
+
+    @TestTemplate
+    public void testRedisWithOutputKeyToConsole(TestContainer container)
+            throws IOException, InterruptedException {
+        Container.ExecResult execResult = container.executeJob("/redis-to-console.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(100, jedis.llen("key_list"));
+        List<String> keyList = jedis.lrange("key_list", 0, 100);
+        for (String key : keyList) {
+            //            each content should have a new column called "origin_key"
+            log.info(key);
+            Assertions.assertTrue(key.contains("origin_key"));
+        }
+        // Clear data to prevent data duplication in the next TestContainer
+        jedis.del("key_list");
+        Assertions.assertEquals(0, jedis.llen("key_list"));
+    }
+
     @TestTemplate
     public void testRedisWithExpire(TestContainer container)
             throws IOException, InterruptedException {