[incubator-inlong] branch master updated: [INLONG-714] Enable checkpointing in inlong-sort

[dockerzhang]

This is an automated email from the ASF dual-hosted git repository.

dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new 5169f56  [INLONG-714] Enable checkpointing in inlong-sort
5169f56 is described below

commit 5169f563724a22d6230805526c3caffbe0dd3229
Author: tianqiwan <tianqi...@tencent.com>
AuthorDate: Sat Jul 10 22:38:17 2021 +0800

    [INLONG-714] Enable checkpointing in inlong-sort
---
 .../org/apache/inlong/sort/configuration/Constants.java   | 15 +++++++++++++++
 .../main/java/org/apache/inlong/sort/flink/Entrance.java  |  5 +++++
 2 files changed, 20 insertions(+)

diff --git 
a/inlong-sort/common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
 
b/inlong-sort/common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
index 4c6c67e..7ecbae2 100644
--- 
a/inlong-sort/common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
+++ 
b/inlong-sort/common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
@@ -226,4 +226,19 @@ public class Constants {
                             + "This controls the frequency to check whether a 
part file should rollover based on"
                             + " 'sink.rolling-policy.rollover-interval'.");
 
+    // ------------------------------------------------------------------------
+    //  Checkpoint related configs
+    // ------------------------------------------------------------------------
+    public static final ConfigOption<Integer> CHECKPOINT_INTERVAL_MS =
+            key("checkpoint.interval")
+                    .defaultValue(600000)
+                    .withDescription("The interval between tow checkpoints");
+
+    public static final ConfigOption<Integer> MIN_PAUSE_BETWEEN_CHECKPOINTS_MS 
=
+            key("min.pause.between.checkpoints.ms")
+                    .defaultValue(500);
+
+    public static final ConfigOption<Integer> CHECKPOINT_TIMEOUT_MS =
+        key("checkpoint.timeout.ms")
+            .defaultValue(600000);
 }
diff --git 
a/inlong-sort/core/src/main/java/org/apache/inlong/sort/flink/Entrance.java 
b/inlong-sort/core/src/main/java/org/apache/inlong/sort/flink/Entrance.java
index bc8cab7..e7ee47b 100644
--- a/inlong-sort/core/src/main/java/org/apache/inlong/sort/flink/Entrance.java
+++ b/inlong-sort/core/src/main/java/org/apache/inlong/sort/flink/Entrance.java
@@ -52,6 +52,11 @@ public class Entrance {
 
         final StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+        
env.enableCheckpointing(config.getInteger(Constants.CHECKPOINT_INTERVAL_MS));
+        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(
+                config.getInteger(Constants.MIN_PAUSE_BETWEEN_CHECKPOINTS_MS));
+        
env.getCheckpointConfig().setCheckpointTimeout(config.getInteger(Constants.CHECKPOINT_TIMEOUT_MS));
+        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
 
         DataStream<SerializedRecord> sourceStream;
         if (sourceType.equals(Constants.SOURCE_TYPE_TUBE)) {