You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2020/08/04 08:01:13 UTC
[flink] 02/02: [FLINK-18748][Runtime/Checkpointing] remove function
and add unit test for NonForced savepoints
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3835007356c61c1967a390b5c3597ac5e1a13374
Author: DG-Wangtao <ta...@deepglint.com>
AuthorDate: Mon Aug 3 23:22:33 2020 +0800
[FLINK-18748][Runtime/Checkpointing] remove function and add unit test for NonForced savepoints
---
.../checkpoint/CheckpointRequestDecider.java | 23 ++++---------
.../checkpoint/CheckpointRequestDeciderTest.java | 40 ++++++++++++++++++++++
2 files changed, 47 insertions(+), 16 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDecider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDecider.java
index d054561..6a9f779 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDecider.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDecider.java
@@ -137,27 +137,18 @@ class CheckpointRequestDecider {
}
CheckpointTriggerRequest first = queuedRequests.first();
- if (first.isForce() || !first.isPeriodic) {
- return Optional.of(queuedRequests.pollFirst());
- }
-
- long nextTriggerDelayMillis = nextTriggerDelayMillis(lastCompletionMs);
- if (nextTriggerDelayMillis > 0) {
- return onTooEarly(nextTriggerDelayMillis);
+ if (!first.isForce() && first.isPeriodic) {
+ long nextTriggerDelayMillis = nextTriggerDelayMillis(lastCompletionMs);
+ if (nextTriggerDelayMillis > 0) {
+ queuedRequests.pollFirst().completeExceptionally(new CheckpointException(MINIMUM_TIME_BETWEEN_CHECKPOINTS));
+ rescheduleTrigger.accept(nextTriggerDelayMillis);
+ return Optional.empty();
+ }
}
return Optional.of(queuedRequests.pollFirst());
}
- private Optional<CheckpointTriggerRequest> onTooEarly(long nextTriggerDelayMillis) {
- CheckpointTriggerRequest first = queuedRequests.first();
- if (first.isPeriodic) {
- queuedRequests.pollFirst().completeExceptionally(new CheckpointException(MINIMUM_TIME_BETWEEN_CHECKPOINTS));
- rescheduleTrigger.accept(nextTriggerDelayMillis);
- }
- return Optional.empty();
- }
-
private long nextTriggerDelayMillis(long lastCheckpointCompletionRelativeTime) {
return lastCheckpointCompletionRelativeTime - clock.relativeTimeMillis() + minPauseBetweenCheckpoints;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDeciderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDeciderTest.java
index f2a3536..55b19b6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDeciderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointRequestDeciderTest.java
@@ -66,6 +66,21 @@ public class CheckpointRequestDeciderTest {
assertEquals(Optional.of(request), decider.chooseQueuedRequestToExecute(isTriggering, 0));
}
+
+ @Test
+ public void testNonForcedEnqueueOnTooManyPending() {
+ final int maxPending = 1;
+ final boolean isTriggering = false;
+ final AtomicInteger currentPending = new AtomicInteger(maxPending);
+ CheckpointRequestDecider decider = decider(Integer.MAX_VALUE, maxPending, 1, currentPending);
+
+ CheckpointTriggerRequest request = nonForcedSavepoint();
+ assertFalse(decider.chooseRequestToExecute(request, isTriggering, 0).isPresent());
+
+ currentPending.set(0);
+ assertEquals(Optional.of(request), decider.chooseQueuedRequestToExecute(isTriggering, 0));
+ }
+
@Test
public void testUserSubmittedPrioritized() {
CheckpointTriggerRequest userSubmitted = regularSavepoint();
@@ -84,6 +99,27 @@ public class CheckpointRequestDeciderTest {
new CheckpointTriggerRequest[]{savepoint, checkpoint});
}
+
+ @Test
+ public void testNonForcedUserSubmittedPrioritized() {
+ CheckpointTriggerRequest userSubmitted = nonForcedSavepoint();
+ CheckpointTriggerRequest periodic = nonForcedPeriodicSavepoint();
+ testRequestsOrdering(
+ new CheckpointTriggerRequest[]{periodic, userSubmitted},
+ new CheckpointTriggerRequest[]{userSubmitted, periodic});
+ }
+
+ @Test
+ public void testNonForcedSavepointPrioritized() {
+ CheckpointTriggerRequest savepoint = nonForcedSavepoint();
+ CheckpointTriggerRequest checkpoint = regularCheckpoint();
+ testRequestsOrdering(
+ new CheckpointTriggerRequest[]{checkpoint, savepoint},
+ new CheckpointTriggerRequest[]{savepoint, checkpoint});
+ }
+
+
+
@Test
public void testQueueSizeLimit() {
final int maxQueuedRequests = 10;
@@ -206,6 +242,10 @@ public class CheckpointRequestDeciderTest {
return savepointRequest(true, true);
}
+ private static CheckpointTriggerRequest nonForcedPeriodicSavepoint(){
+ return savepointRequest(false, true);
+ }
+
private static CheckpointTriggerRequest nonForcedSavepoint() {
return savepointRequest(false, false);
}