remove max-concurrent

apache · Aug 10, 2023 · e4d747e · e4d747e
1 parent 67a603c
commit e4d747e
Show file tree

Hide file tree

Showing 14 changed files with 0 additions and 47 deletions.
diff --git a/config/seatunnel.yaml b/config/seatunnel.yaml
@@ -26,7 +26,6 @@ seatunnel:
     checkpoint:
       interval: 10000
       timeout: 60000
-      max-concurrent: 1
       storage:
         type: hdfs
         max-retained: 3

diff --git a/docs/en/seatunnel-engine/checkpoint-storage.md b/docs/en/seatunnel-engine/checkpoint-storage.md
@@ -59,7 +59,6 @@ seatunnel:
     checkpoint:
       interval: 6000
       timeout: 7000
-      max-concurrent: 1
       storage:
         type: hdfs
         max-retained: 3
@@ -93,7 +92,6 @@ seatunnel:
         checkpoint:
             interval: 6000
             timeout: 7000
-            max-concurrent: 1
             storage:
                 type: hdfs
                 max-retained: 3
@@ -117,7 +115,6 @@ seatunnel:
     checkpoint:
       interval: 6000
       timeout: 7000
-      max-concurrent: 1
       storage:
         type: hdfs
         max-retained: 3
@@ -157,7 +154,6 @@ seatunnel:
     checkpoint:
       interval: 6000
       timeout: 7000
-      max-concurrent: 1
       storage:
         type: hdfs
         max-retained: 3

diff --git a/docs/en/seatunnel-engine/deployment.md b/docs/en/seatunnel-engine/deployment.md
@@ -75,10 +75,6 @@ The interval between two checkpoints, unit is milliseconds. If the `checkpoint.i
 
 The timeout of a checkpoint. If a checkpoint cannot be completed within the timeout period, a checkpoint failure will be triggered. Therefore, Job will be restored.
 
-**max-concurrent**
-
-How many checkpoints can be performed simultaneously at most.
-
 Example
 
 ```
@@ -91,7 +87,6 @@ seatunnel:
         checkpoint:
             interval: 300000
             timeout: 10000
-            max-concurrent: 1
 ```
 
 **checkpoint storage**

diff --git a/...l-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/seatunnel.yaml b/...l-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/seatunnel.yaml
@@ -25,7 +25,6 @@ seatunnel:
     checkpoint:
       interval: 300000
       timeout: 10000
-      max-concurrent: 1
       storage:
         type: localfile
         max-retained: 3

diff --git a/seatunnel-engine/seatunnel-engine-client/src/test/resources/seatunnel.yaml b/seatunnel-engine/seatunnel-engine-client/src/test/resources/seatunnel.yaml
@@ -24,7 +24,6 @@ seatunnel:
         checkpoint:
             interval: 6000
             timeout: 7000
-            max-concurrent: 1
             storage:
                 type: hdfs
                 max-retained: 3

diff --git a/.../main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java b/.../main/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelDomConfigProcessor.java
@@ -156,11 +156,6 @@ private CheckpointConfig parseCheckpointConfig(Node checkpointNode) {
                         getIntegerValue(
                                 ServerConfigOptions.SCHEMA_CHANGE_CHECKPOINT_TIMEOUT.key(),
                                 getTextContent(node)));
-            } else if (ServerConfigOptions.CHECKPOINT_MAX_CONCURRENT.key().equals(name)) {
-                checkpointConfig.setMaxConcurrentCheckpoints(
-                        getIntegerValue(
-                                ServerConfigOptions.CHECKPOINT_MAX_CONCURRENT.key(),
-                                getTextContent(node)));
             } else if (ServerConfigOptions.CHECKPOINT_STORAGE.key().equals(name)) {
                 checkpointConfig.setStorage(parseCheckpointStorageConfig(node));
             } else {

diff --git a/...mmon/src/main/java/org/apache/seatunnel/engine/common/config/server/CheckpointConfig.java b/...mmon/src/main/java/org/apache/seatunnel/engine/common/config/server/CheckpointConfig.java
@@ -33,8 +33,6 @@ public class CheckpointConfig implements Serializable {
     private long checkpointTimeout = ServerConfigOptions.CHECKPOINT_TIMEOUT.defaultValue();
     private long schemaChangeCheckpointTimeout =
             ServerConfigOptions.SCHEMA_CHANGE_CHECKPOINT_TIMEOUT.defaultValue();
-    private int maxConcurrentCheckpoints =
-            ServerConfigOptions.CHECKPOINT_MAX_CONCURRENT.defaultValue();
 
     private CheckpointStorageConfig storage = ServerConfigOptions.CHECKPOINT_STORAGE.defaultValue();
 
@@ -58,11 +56,4 @@ public void setSchemaChangeCheckpointTimeout(long checkpointTimeout) {
                 "The minimum checkpoint timeout is 10 ms.");
         this.schemaChangeCheckpointTimeout = checkpointTimeout;
     }
-
-    public void setMaxConcurrentCheckpoints(int maxConcurrentCheckpoints) {
-        checkArgument(
-                maxConcurrentCheckpoints >= 1,
-                "The minimum number of concurrent checkpoints is 1.");
-        this.maxConcurrentCheckpoints = maxConcurrentCheckpoints;
-    }
 }
diff --git a/...n/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java b/...n/src/main/java/org/apache/seatunnel/engine/common/config/server/ServerConfigOptions.java
@@ -92,12 +92,6 @@ public class ServerConfigOptions {
                     .withDescription(
                             "The timeout (in milliseconds) for a schema change checkpoint.");
 
-    public static final Option<Integer> CHECKPOINT_MAX_CONCURRENT =
-            Options.key("max-concurrent")
-                    .intType()
-                    .defaultValue(1)
-                    .withDescription("The maximum number of concurrent checkpoints.");
-
     public static final Option<String> CHECKPOINT_STORAGE_TYPE =
             Options.key("type")
                     .stringType()

diff --git a/seatunnel-engine/seatunnel-engine-common/src/main/resources/seatunnel.yaml b/seatunnel-engine/seatunnel-engine-common/src/main/resources/seatunnel.yaml
@@ -25,7 +25,6 @@ seatunnel:
         checkpoint:
             interval: 300000
             timeout: 10000
-            max-concurrent: 1
             storage:
                 type: hdfs
                 max-retained: 3

diff --git a/...rc/test/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelConfigParserTest.java b/...rc/test/java/org/apache/seatunnel/engine/common/config/YamlSeaTunnelConfigParserTest.java
@@ -53,9 +53,6 @@ public void testSeaTunnelConfig() {
         Assertions.assertEquals(
                 7000, config.getEngineConfig().getCheckpointConfig().getCheckpointTimeout());
 
-        Assertions.assertEquals(
-                1, config.getEngineConfig().getCheckpointConfig().getMaxConcurrentCheckpoints());
-
         Assertions.assertEquals(
                 "hdfs", config.getEngineConfig().getCheckpointConfig().getStorage().getStorage());
 

diff --git a/seatunnel-engine/seatunnel-engine-common/src/test/resources/seatunnel.yaml b/seatunnel-engine/seatunnel-engine-common/src/test/resources/seatunnel.yaml
@@ -25,7 +25,6 @@ seatunnel:
         checkpoint:
             interval: 6000
             timeout: 7000
-            max-concurrent: 1
             storage:
                 type: hdfs
                 max-retained: 3

diff --git a/...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java b/...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
@@ -390,7 +390,6 @@ protected void tryTriggerPendingCheckpoint(CheckpointType checkpointType) {
         if (checkpointType.notFinalCheckpoint() && checkpointType.notSchemaChangeCheckpoint()) {
             if (currentTimestamp - latestTriggerTimestamp.get()
                             < coordinatorConfig.getCheckpointInterval()
-                    || pendingCounter.get() >= coordinatorConfig.getMaxConcurrentCheckpoints()
                     || !isAllTaskReady) {
                 return;
             }
@@ -737,12 +736,6 @@ public synchronized void completePendingCheckpoint(CompletedCheckpoint completed
         notifyCompleted(completedCheckpoint);
         pendingCheckpoints.remove(checkpointId);
         pendingCounter.decrementAndGet();
-        if (pendingCheckpoints.size() + 1 == coordinatorConfig.getMaxConcurrentCheckpoints()) {
-            // latest checkpoint completed time > checkpoint interval
-            if (completedCheckpoint.getCheckpointType().notFinalCheckpoint()) {
-                scheduleTriggerPendingCheckpoint(0L);
-            }
-        }
         if (isCompleted()) {
             cleanPendingCheckpoint(CheckpointCloseReason.CHECKPOINT_COORDINATOR_COMPLETED);
             if (latestCompletedCheckpoint.getCheckpointType().isSavepoint()) {

diff --git a/...nnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java b/...nnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java
@@ -264,8 +264,6 @@ private CheckpointConfig createJobCheckpointConfig(
         CheckpointConfig jobCheckpointConfig = new CheckpointConfig();
         jobCheckpointConfig.setCheckpointTimeout(defaultCheckpointConfig.getCheckpointTimeout());
         jobCheckpointConfig.setCheckpointInterval(defaultCheckpointConfig.getCheckpointInterval());
-        jobCheckpointConfig.setMaxConcurrentCheckpoints(
-                defaultCheckpointConfig.getMaxConcurrentCheckpoints());
 
         CheckpointStorageConfig jobCheckpointStorageConfig = new CheckpointStorageConfig();
         jobCheckpointStorageConfig.setStorage(defaultCheckpointConfig.getStorage().getStorage());

diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/resources/seatunnel.yaml b/seatunnel-engine/seatunnel-engine-server/src/test/resources/seatunnel.yaml
@@ -25,7 +25,6 @@ seatunnel:
         checkpoint:
             interval: 6000
             timeout: 7000
-            max-concurrent: 1
             storage:
                 type: hdfs
                 max-retained: 3