[Feature][Redis] Flush data when the time reaches checkpoint.interval and update test case (#8308)

Author: lm-ylj

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

@@ -39,6 +39,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 public class RedisSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
         implements SupportMultiTableSinkWriter<Void> {
@@ -78,8 +79,7 @@ public void write(SeaTunnelRow element) throws IOException {
         String value = getValue(element, fields);
         valueBuffer.add(value);
         if (keyBuffer.size() >= batchSize) {
-            doBatchWrite();
-            clearBuffer();
+            flush();
         }
     }
 
@@ -221,6 +221,16 @@ private void doBatchWrite() {
 
     @Override
     public void close() throws IOException {
+        flush();
+    }
+
+    @Override
+    public Optional<Void> prepareCommit() {
+        flush();
+        return Optional.empty();
+    }
+
+    private synchronized void flush() {
         if (!keyBuffer.isEmpty()) {
             doBatchWrite();
             clearBuffer();

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

@@ -16,6 +16,8 @@
  */
 package org.apache.seatunnel.e2e.connector.redis;
 
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
+
 import org.apache.seatunnel.api.table.type.ArrayType;
 import org.apache.seatunnel.api.table.type.BasicType;
 import org.apache.seatunnel.api.table.type.DecimalType;
@@ -25,15 +27,20 @@
 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.utils.JsonUtils;
 import org.apache.seatunnel.e2e.common.TestResource;
 import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.EngineType;
 import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
 import org.apache.seatunnel.format.json.JsonSerializationSchema;
 
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
 import org.testcontainers.containers.Container;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
@@ -52,13 +59,21 @@
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.awaitility.Awaitility.await;
+
 @Slf4j
 public abstract class RedisTestCaseTemplateIT extends TestSuiteBase implements TestResource {
 
@@ -492,7 +507,7 @@ public void testFakeToRedisDeleteSetTest(TestContainer container)
     }
 
     @TestTemplate
-    public void testMysqlCdcToRedisDeleteZSetTest(TestContainer container)
+    public void testFakeToToRedisDeleteZSetTest(TestContainer container)
             throws IOException, InterruptedException {
         Container.ExecResult execResult =
                 container.executeJob("/fake-to-redis-test-delete-zset.conf");
@@ -501,6 +516,69 @@ public void testMysqlCdcToRedisDeleteZSetTest(TestContainer container)
         jedis.del("zset_check");
     }
 
+    @TestTemplate
+    @DisabledOnContainer(
+            value = {},
+            type = {EngineType.SPARK, EngineType.FLINK},
+            disabledReason = "Only support for seatunnel")
+    @DisabledOnOs(OS.WINDOWS)
+    public void testFakeToRedisInRealTimeTest(TestContainer container)
+            throws IOException, InterruptedException {
+        CompletableFuture.supplyAsync(
+                () -> {
+                    try {
+                        container.executeJob("/fake-to-redis-test-in-real-time.conf");
+                    } catch (Exception e) {
+                        log.error("Commit task exception :" + e.getMessage());
+                        throw new RuntimeException(e);
+                    }
+                    return null;
+                });
+        await().atMost(60000, TimeUnit.MILLISECONDS)
+                .untilAsserted(
+                        () -> {
+                            Assertions.assertEquals(3, jedis.llen("list_check"));
+                        });
+        jedis.del("list_check");
+        // Get the task id
+        Container.ExecResult execResult = container.executeBaseCommand(new String[] {"-l"});
+        String regex = "(\\d+)\\s+";
+        Pattern pattern = Pattern.compile(regex);
+        List<String> runningJobId =
+                Arrays.stream(execResult.getStdout().toString().split("\n"))
+                        .filter(s -> s.contains("fake-to-redis-test-in-real-time"))
+                        .map(
+                                s -> {
+                                    Matcher matcher = pattern.matcher(s);
+                                    return matcher.find() ? matcher.group(1) : null;
+                                })
+                        .filter(jobId -> jobId != null)
+                        .collect(Collectors.toList());
+        Assertions.assertEquals(1, runningJobId.size());
+        // Verify that the status is Running
+        for (String jobId : runningJobId) {
+            Container.ExecResult execResult1 =
+                    container.executeBaseCommand(new String[] {"-j", jobId});
+            String stdout = execResult1.getStdout();
+            ObjectNode jsonNodes = JsonUtils.parseObject(stdout);
+            Assertions.assertEquals(jsonNodes.get("jobStatus").asText(), "RUNNING");
+        }
+        // Execute cancellation task
+        String[] batchCancelCommand =
+                Stream.concat(Arrays.stream(new String[] {"-can"}), runningJobId.stream())
+                        .toArray(String[]::new);
+        Assertions.assertEquals(0, container.executeBaseCommand(batchCancelCommand).getExitCode());
+
+        // Verify whether the cancellation is successful
+        for (String jobId : runningJobId) {
+            Container.ExecResult execResult1 =
+                    container.executeBaseCommand(new String[] {"-j", jobId});
+            String stdout = execResult1.getStdout();
+            ObjectNode jsonNodes = JsonUtils.parseObject(stdout);
+            Assertions.assertEquals(jsonNodes.get("jobStatus").asText(), "CANCELED");
+        }
+    }
+
     @TestTemplate
     public void testFakeToRedisNormalKeyIsNullTest(TestContainer container)
             throws IOException, InterruptedException {

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/fake-to-redis-test-in-real-time.conf

@@ -0,0 +1,68 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 10000
+  shade.identifier = "base64"
+}
+
+source {
+  FakeSource {
+    schema = {
+      fields {
+                id = int
+                val_bool = boolean
+                val_int8 = tinyint
+                val_int16 = smallint
+                val_int32 = int
+                val_int64 = bigint
+                val_float = float
+                val_double = double
+                val_decimal = "decimal(16, 1)"
+                val_string = string
+                val_unixtime_micros = timestamp
+      }
+    }
+    rows = [
+      {
+        kind = INSERT
+        fields = [1, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
+      },
+      {
+        kind = INSERT
+        fields = [2, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
+      },
+      {
+        kind = INSERT
+        fields = [3, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW", "2020-02-02T02:02:02"]
+      }
+    ]
+  }
+}
+
+sink {
+  Redis {
+    host = "redis-e2e"
+    port = 6379
+    auth = "U2VhVHVubmVs"
+    key = "list_check"
+    data_type = list
+    batch_size = 33
+  }
+}