You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2021/06/29 07:29:27 UTC
[flink] branch master updated (6e62a41 -> 9d449ff)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 6e62a41 [hotfix][docs] fix typos (#16300)
new 2af2740 [FLINK-23041][streaming] Calculation of timeout for switching from AC to UC is based on the global checkpoint start time rather than the local first barrier received time
new 5040eaa [FLINK-23041][runtime] Added new well defined checkpoint configuration aligned-checkpoint-timeout instead of alignment-timeout
new 9d449ff [FLINK-23041][webui] Renamed alignment_timeout to aligned_checkpoint_timeout in WebUI
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../docs/ops/state/unaligned_checkpoints.md | 14 ++--
.../execution_checkpointing_configuration.html | 2 +-
.../shortcodes/generated/rest_v1_dispatcher.html | 6 +-
.../flink/state/api/output/SnapshotUtils.java | 2 +-
.../src/test/resources/rest_api_v1.snapshot | 2 +-
.../src/app/interfaces/job-checkpoint.ts | 2 +-
.../job/checkpoints/job-checkpoints.component.html | 4 +-
.../runtime/checkpoint/CheckpointCoordinator.java | 6 +-
.../runtime/checkpoint/CheckpointOptions.java | 61 ++++++++++--------
.../network/api/serialization/EventSerializer.java | 2 +-
.../tasks/CheckpointCoordinatorConfiguration.java | 24 ++++---
.../job/checkpoints/CheckpointConfigHandler.java | 2 +-
.../messages/checkpoints/CheckpointConfigInfo.java | 14 ++--
.../checkpoint/CheckpointCoordinatorTest.java | 4 +-
.../runtime/checkpoint/CheckpointOptionsTest.java | 19 +++---
.../api/environment/CheckpointConfig.java | 58 +++++++++++++----
.../environment/ExecutionCheckpointingOptions.java | 27 +++++++-
.../flink/streaming/api/graph/StreamConfig.java | 9 +--
.../api/graph/StreamingJobGraphGenerator.java | 5 +-
.../SingleCheckpointBarrierHandler.java | 16 +++--
.../runtime/tasks/SourceOperatorStreamTask.java | 2 +-
.../streaming/runtime/tasks/SourceStreamTask.java | 2 +-
.../checkpointing/AlternatingCheckpointsTest.java | 74 ++++++++++++++++++++--
...tStreamTaskChainedSourcesCheckpointingTest.java | 2 +-
.../runtime/tasks/MultipleInputStreamTaskTest.java | 2 +-
25 files changed, 253 insertions(+), 108 deletions(-)
[flink] 01/03: [FLINK-23041][streaming] Calculation of timeout for
switching from AC to UC is based on the global checkpoint start time rather
than the local first barrier received time
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2af2740bd5251b7cc39a56f9b73ace1f1cfbb871
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Tue Jun 22 20:16:33 2021 +0300
[FLINK-23041][streaming] Calculation of timeout for switching from AC to UC is based on the global checkpoint start time rather than the local first barrier received time
---
.../SingleCheckpointBarrierHandler.java | 16 +++--
.../checkpointing/AlternatingCheckpointsTest.java | 74 ++++++++++++++++++++--
2 files changed, 80 insertions(+), 10 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
index 83c324c..fcca333 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
@@ -82,7 +82,6 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
private CompletableFuture<Void> allBarriersReceivedFuture = new CompletableFuture<>();
private BarrierHandlerState currentState;
- private long firstBarrierArrivalTime;
private Cancellable currentAlignmentTimer;
private final boolean alternating;
@@ -271,6 +270,13 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
}
private void registerAlignmentTimer(CheckpointBarrier announcedBarrier) {
+ long alignedCheckpointTimeout =
+ announcedBarrier.getCheckpointOptions().getAlignmentTimeout();
+ long timePassedSinceCheckpointStart =
+ getClock().absoluteTimeMillis() - announcedBarrier.getTimestamp();
+
+ long timerDelay = Math.max(alignedCheckpointTimeout - timePassedSinceCheckpointStart, 0);
+
this.currentAlignmentTimer =
registerTimer.apply(
() -> {
@@ -290,8 +296,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
currentAlignmentTimer = null;
return null;
},
- Duration.ofMillis(
- announcedBarrier.getCheckpointOptions().getAlignmentTimeout()));
+ Duration.ofMillis(timerDelay));
}
private void checkNewCheckpoint(CheckpointBarrier barrier) throws IOException {
@@ -306,7 +311,6 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
currentCheckpointId = barrierId;
numBarriersReceived = 0;
allBarriersReceivedFuture = new CompletableFuture<>();
- firstBarrierArrivalTime = getClock().relativeTimeNanos();
if (alternating && barrier.getCheckpointOptions().isTimeoutable()) {
registerAlignmentTimer(barrier);
@@ -432,8 +436,8 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
public boolean isTimedOut(CheckpointBarrier barrier) {
return barrier.getCheckpointOptions().isTimeoutable()
&& barrier.getId() <= currentCheckpointId
- && barrier.getCheckpointOptions().getAlignmentTimeout() * 1_000_000
- < (getClock().relativeTimeNanos() - firstBarrierArrivalTime);
+ && barrier.getCheckpointOptions().getAlignmentTimeout()
+ < (getClock().absoluteTimeMillis() - barrier.getTimestamp());
}
@Override
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCheckpointsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCheckpointsTest.java
index d4acc1f..456e8af 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCheckpointsTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCheckpointsTest.java
@@ -354,6 +354,30 @@ public class AlternatingCheckpointsTest {
}
@Test
+ public void testTimeoutAlignmentBeforeFirstBarrier() throws Exception {
+ // given: Local channels.
+ int numChannels = 2;
+ ValidatingCheckpointHandler target = new ValidatingCheckpointHandler();
+ CheckpointedInputGate gate =
+ new TestCheckpointedInputGateBuilder(
+ numChannels, getTestBarrierHandlerFactory(target))
+ .withTestChannels()
+ .withMailboxExecutor()
+ .build();
+
+ long alignedCheckpointTimeout = 100;
+ // when: Aligned checkpoint timeout expired before the first barrier received.
+ Buffer checkpointBarrier = withTimeout(1, alignedCheckpointTimeout);
+ clock.advanceTime(alignedCheckpointTimeout + 1, TimeUnit.MILLISECONDS);
+
+ ((TestInputChannel) gate.getChannel(0)).read(checkpointBarrier.retainBuffer());
+
+ // then: The UC is triggered as soon as the first barrier is received.
+ assertBarrier(gate);
+ assertEquals(1, target.getTriggeredCheckpointCounter());
+ }
+
+ @Test
public void testTimeoutAlignmentWhenLocalBarrierFirst() throws Exception {
// given: Gate with remote and local channels.
int numChannels = 3;
@@ -737,6 +761,48 @@ public class AlternatingCheckpointsTest {
}
@Test
+ public void testActiveTimeoutBeforeFirstAnnouncementPassiveTimeout() throws Exception {
+ // given: Two barriers from two channels.
+ int numChannels = 2;
+ ValidatingCheckpointHandler target = new ValidatingCheckpointHandler();
+ try (CheckpointedInputGate gate =
+ new TestCheckpointedInputGateBuilder(
+ numChannels, getTestBarrierHandlerFactory(target))
+ .withRemoteChannels()
+ .withMailboxExecutor()
+ .build()) {
+ long alignmentCheckpointTimeout = 10;
+ Buffer checkpointBarrier = withTimeout(alignmentCheckpointTimeout);
+
+ getChannel(gate, 0).onBuffer(dataBuffer(), 0, 0);
+ getChannel(gate, 0).onBuffer(checkpointBarrier.retainBuffer(), 1, 0);
+ getChannel(gate, 1).onBuffer(dataBuffer(), 0, 0);
+ getChannel(gate, 1).onBuffer(checkpointBarrier.retainBuffer(), 1, 0);
+
+ assertEquals(0, target.getTriggeredCheckpointCounter());
+
+ // when: The receiving of the first announcement is delayed on more than alignment
+ // checkpoint timeout.
+ clock.advanceTimeWithoutRunningCallables(
+ alignmentCheckpointTimeout + 1, TimeUnit.MILLISECONDS);
+
+ assertAnnouncement(gate);
+ // we simulate active time out firing after the passive one
+ clock.executeCallables();
+
+ // then: Barriers should be reprioritized and the UC should be triggered.
+ assertAnnouncement(gate);
+ assertBarrier(gate);
+ assertBarrier(gate);
+ assertEquals(1, target.getTriggeredCheckpointCounter());
+ assertThat(target.getTriggeredCheckpointOptions(), contains(unaligned(getDefault())));
+ // Followed by overtaken buffers
+ assertData(gate);
+ assertData(gate);
+ }
+ }
+
+ @Test
public void testActiveTimeoutAfterBarrierPassiveTimeout() throws Exception {
int numChannels = 2;
ValidatingCheckpointHandler target = new ValidatingCheckpointHandler();
@@ -1112,15 +1178,15 @@ public class AlternatingCheckpointsTest {
while (checkpointedGate.pollNext().isPresent()) {}
}
- private Buffer withTimeout(long alignmentTimeout) throws IOException {
- return withTimeout(1, alignmentTimeout);
+ private Buffer withTimeout(long alignedCheckpointTimeout) throws IOException {
+ return withTimeout(1, alignedCheckpointTimeout);
}
- private Buffer withTimeout(int checkpointId, long alignmentTimeout) throws IOException {
+ private Buffer withTimeout(int checkpointId, long alignedCheckpointTimeout) throws IOException {
return barrier(
checkpointId,
clock.relativeTimeMillis(),
- alignedWithTimeout(getDefault(), alignmentTimeout));
+ alignedWithTimeout(getDefault(), alignedCheckpointTimeout));
}
private Buffer barrier(long barrierId, long barrierTimestamp, CheckpointOptions options)
[flink] 03/03: [FLINK-23041][webui] Renamed alignment_timeout to
aligned_checkpoint_timeout in WebUI
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9d449ff540b5e0d4e97d8a0377438f8ed1b70a6e
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Tue Jun 22 20:18:45 2021 +0300
[FLINK-23041][webui] Renamed alignment_timeout to aligned_checkpoint_timeout in WebUI
This closes #16227
---
docs/content/docs/ops/state/unaligned_checkpoints.md | 14 ++++++++------
.../generated/execution_checkpointing_configuration.html | 2 +-
docs/layouts/shortcodes/generated/rest_v1_dispatcher.html | 6 +++---
flink-runtime-web/src/test/resources/rest_api_v1.snapshot | 2 +-
.../web-dashboard/src/app/interfaces/job-checkpoint.ts | 2 +-
.../pages/job/checkpoints/job-checkpoints.component.html | 4 ++--
.../rest/messages/checkpoints/CheckpointConfigInfo.java | 14 +++++++-------
7 files changed, 23 insertions(+), 21 deletions(-)
diff --git a/docs/content/docs/ops/state/unaligned_checkpoints.md b/docs/content/docs/ops/state/unaligned_checkpoints.md
index 01ae924..b47e43e 100644
--- a/docs/content/docs/ops/state/unaligned_checkpoints.md
+++ b/docs/content/docs/ops/state/unaligned_checkpoints.md
@@ -37,23 +37,25 @@ backpressure. Then, checkpointing time becomes mostly independent of the end-to-
aware unaligned checkpointing adds to I/O to the state storage, so you shouldn't use it when the
I/O to the state storage is actually the bottleneck during checkpointing.
-### Alignment timeout
+### Aligned checkpoint timeout
-After enabling unaligned checkpoints, you can also specify the alignment timeout programmatically:
+After enabling unaligned checkpoints, you can also specify the aligned checkpoint timeout
+programmatically:
```java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-env.getCheckpointConfig().setAlignmentTimeout(Duration.ofSeconds(30));
+env.getCheckpointConfig().setAlignedCheckpointTimeout(Duration.ofSeconds(30));
```
or in the `flink-conf.yml` configuration file:
```
-execution.checkpointing.alignment-timeout: 30 s
+execution.checkpointing.aligned-checkpoint-timeout: 30 s
```
-When activated, each checkpoint will still begin as an aligned checkpoint, but if the alignment time
-for some subtask exceeds this timeout, then the checkpoint will proceed as an unaligned checkpoint.
+When activated, each checkpoint will still begin as an aligned checkpoint, but if the time between
+the start of the global checkpoint and the start of the checkpoint on a subtask exceeds the aligned
+checkpoint timeout, then the checkpoint will proceed as an unaligned checkpoint.
## Limitations
diff --git a/docs/layouts/shortcodes/generated/execution_checkpointing_configuration.html b/docs/layouts/shortcodes/generated/execution_checkpointing_configuration.html
index 90b6e9bd..371fc73 100644
--- a/docs/layouts/shortcodes/generated/execution_checkpointing_configuration.html
+++ b/docs/layouts/shortcodes/generated/execution_checkpointing_configuration.html
@@ -9,7 +9,7 @@
</thead>
<tbody>
<tr>
- <td><h5>execution.checkpointing.alignment-timeout</h5></td>
+ <td><h5>execution.checkpointing.aligned-checkpoint-timeout</h5></td>
<td style="word-wrap: break-word;">0 ms</td>
<td>Duration</td>
<td>Only relevant if <code class="highlighter-rouge">execution.checkpointing.unaligned</code> is enabled.<br /><br />If timeout is 0, checkpoints will always start unaligned.<br /><br />If timeout has a positive value, checkpoints will start aligned. If during checkpointing, checkpoint start delay exceeds this timeout, alignment will timeout and checkpoint barrier will start working as unaligned checkpoint.</td>
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index 522eba6..2329226 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -101,10 +101,10 @@
"type" : "object",
"id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:DashboardConfiguration:Features",
"properties" : {
- "web-submit" : {
+ "web-cancel" : {
"type" : "boolean"
},
- "web-cancel" : {
+ "web-submit" : {
"type" : "boolean"
}
}
@@ -2181,7 +2181,7 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"type" : "object",
"id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:checkpoints:CheckpointConfigInfo",
"properties" : {
- "alignment_timeout" : {
+ "aligned_checkpoint_timeout" : {
"type" : "integer"
},
"checkpoint_storage" : {
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index d64a73a..fb7fd70 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -1258,7 +1258,7 @@
"tolerable_failed_checkpoints" : {
"type" : "integer"
},
- "alignment_timeout" : {
+ "aligned_checkpoint_timeout" : {
"type" : "integer"
}
}
diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts
index b850215..7e7c441 100644
--- a/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts
+++ b/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts
@@ -119,7 +119,7 @@ export interface CheckPointConfigInterface {
checkpoint_storage: string;
unaligned_checkpoints: boolean;
tolerable_failed_checkpoints: number;
- alignment_timeout: number;
+ aligned_checkpoint_timeout: number;
}
export interface CheckPointDetailInterface {
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html
index 00cb2c7..e7aeb5b 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html
@@ -253,8 +253,8 @@
<td>{{ checkPointConfig['unaligned_checkpoints'] ? 'Enabled' : 'Disabled' }}</td>
</tr>
<tr *ngIf="checkPointConfig['unaligned_checkpoints']">
- <td>Alignment timeout</td>
- <td>{{ checkPointConfig['alignment_timeout'] | humanizeDuration}}</td>
+ <td>Aligned checkpoint timeout</td>
+ <td>{{ checkPointConfig['aligned_checkpoint_timeout'] | humanizeDuration}}</td>
</tr>
<tr *ngIf="checkPointConfig['externalization']">
<td>Persist Checkpoints Externally</td>
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfo.java
index e5ca260..56598d6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointConfigInfo.java
@@ -60,7 +60,7 @@ public class CheckpointConfigInfo implements ResponseBody {
public static final String FIELD_NAME_TOLERABLE_FAILED_CHECKPOINTS =
"tolerable_failed_checkpoints";
- public static final String FIELD_NAME_ALIGNMENT_TIMEOUT = "alignment_timeout";
+ public static final String FIELD_NAME_ALIGNED_CHECKPOINT_TIMEOUT = "aligned_checkpoint_timeout";
@JsonProperty(FIELD_NAME_PROCESSING_MODE)
private final ProcessingMode processingMode;
@@ -92,8 +92,8 @@ public class CheckpointConfigInfo implements ResponseBody {
@JsonProperty(FIELD_NAME_TOLERABLE_FAILED_CHECKPOINTS)
private final int tolerableFailedCheckpoints;
- @JsonProperty(FIELD_NAME_ALIGNMENT_TIMEOUT)
- private final long alignmentTimeout;
+ @JsonProperty(FIELD_NAME_ALIGNED_CHECKPOINT_TIMEOUT)
+ private final long alignedCheckpointTimeout;
@JsonCreator
public CheckpointConfigInfo(
@@ -108,7 +108,7 @@ public class CheckpointConfigInfo implements ResponseBody {
@JsonProperty(FIELD_NAME_CHECKPOINT_STORAGE) String checkpointStorage,
@JsonProperty(FIELD_NAME_UNALIGNED_CHECKPOINTS) boolean unalignedCheckpoints,
@JsonProperty(FIELD_NAME_TOLERABLE_FAILED_CHECKPOINTS) int tolerableFailedCheckpoints,
- @JsonProperty(FIELD_NAME_ALIGNMENT_TIMEOUT) long alignmentTimeout) {
+ @JsonProperty(FIELD_NAME_ALIGNED_CHECKPOINT_TIMEOUT) long alignedCheckpointTimeout) {
this.processingMode = Preconditions.checkNotNull(processingMode);
this.checkpointInterval = checkpointInterval;
this.checkpointTimeout = checkpointTimeout;
@@ -119,7 +119,7 @@ public class CheckpointConfigInfo implements ResponseBody {
this.checkpointStorage = Preconditions.checkNotNull(checkpointStorage);
this.unalignedCheckpoints = unalignedCheckpoints;
this.tolerableFailedCheckpoints = tolerableFailedCheckpoints;
- this.alignmentTimeout = alignmentTimeout;
+ this.alignedCheckpointTimeout = alignedCheckpointTimeout;
}
@Override
@@ -141,7 +141,7 @@ public class CheckpointConfigInfo implements ResponseBody {
&& Objects.equals(checkpointStorage, that.checkpointStorage)
&& unalignedCheckpoints == that.unalignedCheckpoints
&& tolerableFailedCheckpoints == that.tolerableFailedCheckpoints
- && alignmentTimeout == that.alignmentTimeout;
+ && alignedCheckpointTimeout == that.alignedCheckpointTimeout;
}
@Override
@@ -157,7 +157,7 @@ public class CheckpointConfigInfo implements ResponseBody {
checkpointStorage,
unalignedCheckpoints,
tolerableFailedCheckpoints,
- alignmentTimeout);
+ alignedCheckpointTimeout);
}
/** Contains information about the externalized checkpoint configuration. */
[flink] 02/03: [FLINK-23041][runtime] Added new well defined
checkpoint configuration aligned-checkpoint-timeout instead of
alignment-timeout
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5040eaa877e59e1399cee1cf37ad1d33f62d01da
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Tue Jun 22 20:18:07 2021 +0300
[FLINK-23041][runtime] Added new well defined checkpoint configuration aligned-checkpoint-timeout instead of alignment-timeout
---
.../flink/state/api/output/SnapshotUtils.java | 2 +-
.../runtime/checkpoint/CheckpointCoordinator.java | 6 +--
.../runtime/checkpoint/CheckpointOptions.java | 61 ++++++++++++----------
.../network/api/serialization/EventSerializer.java | 2 +-
.../tasks/CheckpointCoordinatorConfiguration.java | 24 +++++----
.../job/checkpoints/CheckpointConfigHandler.java | 2 +-
.../checkpoint/CheckpointCoordinatorTest.java | 4 +-
.../runtime/checkpoint/CheckpointOptionsTest.java | 19 ++++---
.../api/environment/CheckpointConfig.java | 58 +++++++++++++++-----
.../environment/ExecutionCheckpointingOptions.java | 27 +++++++++-
.../flink/streaming/api/graph/StreamConfig.java | 9 ++--
.../api/graph/StreamingJobGraphGenerator.java | 5 +-
.../SingleCheckpointBarrierHandler.java | 4 +-
.../runtime/tasks/SourceOperatorStreamTask.java | 2 +-
.../streaming/runtime/tasks/SourceStreamTask.java | 2 +-
...tStreamTaskChainedSourcesCheckpointingTest.java | 2 +-
.../runtime/tasks/MultipleInputStreamTaskTest.java | 2 +-
17 files changed, 152 insertions(+), 79 deletions(-)
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SnapshotUtils.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SnapshotUtils.java
index e62e84e..9a1867f 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SnapshotUtils.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SnapshotUtils.java
@@ -53,7 +53,7 @@ public final class SnapshotUtils {
AbstractFsCheckpointStorageAccess.encodePathAsReference(savepointPath),
isExactlyOnceMode,
isUnalignedCheckpoint,
- CheckpointOptions.NO_ALIGNMENT_TIME_OUT);
+ CheckpointOptions.NO_ALIGNED_CHECKPOINT_TIME_OUT);
operator.prepareSnapshotPreBarrier(CHECKPOINT_ID);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index 16d588d..896ed7b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -170,7 +170,7 @@ public class CheckpointCoordinator {
private final boolean unalignedCheckpointsEnabled;
- private final long alignmentTimeout;
+ private final long alignedCheckpointTimeout;
/** Actor that receives status updates from the execution graph this coordinator works for. */
private JobStatusListener jobStatusListener;
@@ -311,7 +311,7 @@ public class CheckpointCoordinator {
this.clock = checkNotNull(clock);
this.isExactlyOnceMode = chkConfig.isExactlyOnce();
this.unalignedCheckpointsEnabled = chkConfig.isUnalignedCheckpointsEnabled();
- this.alignmentTimeout = chkConfig.getAlignmentTimeout();
+ this.alignedCheckpointTimeout = chkConfig.getAlignedCheckpointTimeout();
this.checkpointIdOfIgnoredInFlightData = chkConfig.getCheckpointIdOfIgnoredInFlightData();
this.recentPendingCheckpoints = new ArrayDeque<>(NUM_GHOST_CHECKPOINT_IDS);
@@ -824,7 +824,7 @@ public class CheckpointCoordinator {
checkpointStorageLocation.getLocationReference(),
isExactlyOnceMode,
unalignedCheckpointsEnabled,
- alignmentTimeout);
+ alignedCheckpointTimeout);
// send the messages to the tasks that trigger their checkpoint
for (Execution execution : tasksToTrigger) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointOptions.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointOptions.java
index 8076cb9..3b7e2b6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointOptions.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointOptions.java
@@ -47,7 +47,7 @@ public class CheckpointOptions implements Serializable {
FORCED_ALIGNED
}
- public static final long NO_ALIGNMENT_TIME_OUT = Long.MAX_VALUE;
+ public static final long NO_ALIGNED_CHECKPOINT_TIME_OUT = Long.MAX_VALUE;
private static final long serialVersionUID = 5010126558083292915L;
@@ -59,17 +59,18 @@ public class CheckpointOptions implements Serializable {
private final AlignmentType alignmentType;
- private final long alignmentTimeout;
+ private final long alignedCheckpointTimeout;
public static CheckpointOptions notExactlyOnce(
CheckpointType type, CheckpointStorageLocationReference location) {
return new CheckpointOptions(
- type, location, AlignmentType.AT_LEAST_ONCE, NO_ALIGNMENT_TIME_OUT);
+ type, location, AlignmentType.AT_LEAST_ONCE, NO_ALIGNED_CHECKPOINT_TIME_OUT);
}
public static CheckpointOptions alignedNoTimeout(
CheckpointType type, CheckpointStorageLocationReference location) {
- return new CheckpointOptions(type, location, AlignmentType.ALIGNED, NO_ALIGNMENT_TIME_OUT);
+ return new CheckpointOptions(
+ type, location, AlignmentType.ALIGNED, NO_ALIGNED_CHECKPOINT_TIME_OUT);
}
public static CheckpointOptions unaligned(CheckpointStorageLocationReference location) {
@@ -77,22 +78,25 @@ public class CheckpointOptions implements Serializable {
CheckpointType.CHECKPOINT,
location,
AlignmentType.UNALIGNED,
- NO_ALIGNMENT_TIME_OUT);
+ NO_ALIGNED_CHECKPOINT_TIME_OUT);
}
public static CheckpointOptions alignedWithTimeout(
- CheckpointStorageLocationReference location, long alignmentTimeout) {
+ CheckpointStorageLocationReference location, long alignedCheckpointTimeout) {
return new CheckpointOptions(
- CheckpointType.CHECKPOINT, location, AlignmentType.ALIGNED, alignmentTimeout);
+ CheckpointType.CHECKPOINT,
+ location,
+ AlignmentType.ALIGNED,
+ alignedCheckpointTimeout);
}
private static CheckpointOptions forceAligned(
- CheckpointStorageLocationReference location, long alignmentTimeout) {
+ CheckpointStorageLocationReference location, long alignedCheckpointTimeout) {
return new CheckpointOptions(
CheckpointType.CHECKPOINT,
location,
AlignmentType.FORCED_ALIGNED,
- alignmentTimeout);
+ alignedCheckpointTimeout);
}
public static CheckpointOptions forConfig(
@@ -100,44 +104,45 @@ public class CheckpointOptions implements Serializable {
CheckpointStorageLocationReference locationReference,
boolean isExactlyOnceMode,
boolean isUnalignedEnabled,
- long alignmentTimeout) {
+ long alignedCheckpointTimeout) {
if (!isExactlyOnceMode) {
return notExactlyOnce(checkpointType, locationReference);
} else if (checkpointType.isSavepoint()) {
return alignedNoTimeout(checkpointType, locationReference);
} else if (!isUnalignedEnabled) {
return alignedNoTimeout(checkpointType, locationReference);
- } else if (alignmentTimeout == 0 || alignmentTimeout == NO_ALIGNMENT_TIME_OUT) {
+ } else if (alignedCheckpointTimeout == 0
+ || alignedCheckpointTimeout == NO_ALIGNED_CHECKPOINT_TIME_OUT) {
return unaligned(locationReference);
} else {
- return alignedWithTimeout(locationReference, alignmentTimeout);
+ return alignedWithTimeout(locationReference, alignedCheckpointTimeout);
}
}
@VisibleForTesting
public CheckpointOptions(
CheckpointType checkpointType, CheckpointStorageLocationReference targetLocation) {
- this(checkpointType, targetLocation, AlignmentType.ALIGNED, NO_ALIGNMENT_TIME_OUT);
+ this(checkpointType, targetLocation, AlignmentType.ALIGNED, NO_ALIGNED_CHECKPOINT_TIME_OUT);
}
public CheckpointOptions(
CheckpointType checkpointType,
CheckpointStorageLocationReference targetLocation,
AlignmentType alignmentType,
- long alignmentTimeout) {
+ long alignedCheckpointTimeout) {
checkArgument(
alignmentType != AlignmentType.UNALIGNED || !checkpointType.isSavepoint(),
"Savepoint can't be unaligned");
checkArgument(
- alignmentTimeout == NO_ALIGNMENT_TIME_OUT
+ alignedCheckpointTimeout == NO_ALIGNED_CHECKPOINT_TIME_OUT
|| alignmentType != AlignmentType.UNALIGNED,
"Unaligned checkpoint can't have timeout (%s)",
- alignmentTimeout);
+ alignedCheckpointTimeout);
this.checkpointType = checkNotNull(checkpointType);
this.targetLocation = checkNotNull(targetLocation);
this.alignmentType = checkNotNull(alignmentType);
- this.alignmentTimeout = alignmentTimeout;
+ this.alignedCheckpointTimeout = alignedCheckpointTimeout;
}
public boolean needsAlignment() {
@@ -145,8 +150,8 @@ public class CheckpointOptions implements Serializable {
&& (getCheckpointType().isSavepoint() || !isUnalignedCheckpoint());
}
- public long getAlignmentTimeout() {
- return alignmentTimeout;
+ public long getAlignedCheckpointTimeout() {
+ return alignedCheckpointTimeout;
}
public AlignmentType getAlignment() {
@@ -158,7 +163,8 @@ public class CheckpointOptions implements Serializable {
return false;
}
return alignmentType == AlignmentType.ALIGNED
- && (alignmentTimeout > 0 && alignmentTimeout != NO_ALIGNMENT_TIME_OUT);
+ && (alignedCheckpointTimeout > 0
+ && alignedCheckpointTimeout != NO_ALIGNED_CHECKPOINT_TIME_OUT);
}
// ------------------------------------------------------------------------
@@ -183,8 +189,8 @@ public class CheckpointOptions implements Serializable {
public CheckpointOptions withUnalignedSupported() {
if (alignmentType == AlignmentType.FORCED_ALIGNED) {
- return alignmentTimeout != NO_ALIGNMENT_TIME_OUT
- ? alignedWithTimeout(targetLocation, alignmentTimeout)
+ return alignedCheckpointTimeout != NO_ALIGNED_CHECKPOINT_TIME_OUT
+ ? alignedWithTimeout(targetLocation, alignedCheckpointTimeout)
: unaligned(targetLocation);
}
return this;
@@ -192,7 +198,7 @@ public class CheckpointOptions implements Serializable {
public CheckpointOptions withUnalignedUnsupported() {
if (isUnalignedCheckpoint() || isTimeoutable()) {
- return forceAligned(targetLocation, alignmentTimeout);
+ return forceAligned(targetLocation, alignedCheckpointTimeout);
}
return this;
}
@@ -201,7 +207,8 @@ public class CheckpointOptions implements Serializable {
@Override
public int hashCode() {
- return Objects.hash(targetLocation, checkpointType, alignmentType, alignmentTimeout);
+ return Objects.hash(
+ targetLocation, checkpointType, alignmentType, alignedCheckpointTimeout);
}
@Override
@@ -213,7 +220,7 @@ public class CheckpointOptions implements Serializable {
return this.checkpointType == that.checkpointType
&& this.targetLocation.equals(that.targetLocation)
&& this.alignmentType == that.alignmentType
- && this.alignmentTimeout == that.alignmentTimeout;
+ && this.alignedCheckpointTimeout == that.alignedCheckpointTimeout;
} else {
return false;
}
@@ -228,8 +235,8 @@ public class CheckpointOptions implements Serializable {
+ targetLocation
+ ", alignment = "
+ alignmentType
- + ", alignmentTimeout = "
- + alignmentTimeout
+ + ", alignedCheckpointTimeout = "
+ + alignedCheckpointTimeout
+ "}";
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java
index 54f4a9e..ae85a0b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java
@@ -236,7 +236,7 @@ public class EventSerializer {
buf.put(locationBytes);
}
buf.put((byte) checkpointOptions.getAlignment().ordinal());
- buf.putLong(checkpointOptions.getAlignmentTimeout());
+ buf.putLong(checkpointOptions.getAlignedCheckpointTimeout());
buf.flip();
return buf;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java
index 0fd6f87..1a27bfb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java
@@ -62,7 +62,7 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
private final boolean isUnalignedCheckpointsEnabled;
- private final long alignmentTimeout;
+ private final long alignedCheckpointTimeout;
private final long checkpointIdOfIgnoredInFlightData;
@@ -104,7 +104,7 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
boolean isPreferCheckpointForRecovery,
int tolerableCpFailureNumber,
boolean isUnalignedCheckpointsEnabled,
- long alignmentTimeout,
+ long alignedCheckpointTimeout,
long checkpointIdOfIgnoredInFlightData) {
// sanity checks
@@ -128,7 +128,7 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
this.isPreferCheckpointForRecovery = isPreferCheckpointForRecovery;
this.tolerableCheckpointFailureNumber = tolerableCpFailureNumber;
this.isUnalignedCheckpointsEnabled = isUnalignedCheckpointsEnabled;
- this.alignmentTimeout = alignmentTimeout;
+ this.alignedCheckpointTimeout = alignedCheckpointTimeout;
this.checkpointIdOfIgnoredInFlightData = checkpointIdOfIgnoredInFlightData;
}
@@ -168,8 +168,8 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
return isUnalignedCheckpointsEnabled;
}
- public long getAlignmentTimeout() {
- return alignmentTimeout;
+ public long getAlignedCheckpointTimeout() {
+ return alignedCheckpointTimeout;
}
public long getCheckpointIdOfIgnoredInFlightData() {
@@ -191,6 +191,7 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
&& maxConcurrentCheckpoints == that.maxConcurrentCheckpoints
&& isExactlyOnce == that.isExactlyOnce
&& isUnalignedCheckpointsEnabled == that.isUnalignedCheckpointsEnabled
+ && alignedCheckpointTimeout == that.alignedCheckpointTimeout
&& checkpointRetentionPolicy == that.checkpointRetentionPolicy
&& isPreferCheckpointForRecovery == that.isPreferCheckpointForRecovery
&& tolerableCheckpointFailureNumber == that.tolerableCheckpointFailureNumber
@@ -207,6 +208,7 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
checkpointRetentionPolicy,
isExactlyOnce,
isUnalignedCheckpointsEnabled,
+ alignedCheckpointTimeout,
isPreferCheckpointForRecovery,
tolerableCheckpointFailureNumber,
checkpointIdOfIgnoredInFlightData);
@@ -229,6 +231,8 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
+ isExactlyOnce
+ ", isUnalignedCheckpoint="
+ isUnalignedCheckpointsEnabled
+ + ", alignedCheckpointTimeout="
+ + alignedCheckpointTimeout
+ ", isPreferCheckpointForRecovery="
+ isPreferCheckpointForRecovery
+ ", tolerableCheckpointFailureNumber="
@@ -254,7 +258,7 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
private boolean isPreferCheckpointForRecovery = true;
private int tolerableCheckpointFailureNumber;
private boolean isUnalignedCheckpointsEnabled;
- private long alignmentTimeout = 0;
+ private long alignedCheckpointTimeout = 0;
private long checkpointIdOfIgnoredInFlightData;
public CheckpointCoordinatorConfiguration build() {
@@ -268,7 +272,7 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
isPreferCheckpointForRecovery,
tolerableCheckpointFailureNumber,
isUnalignedCheckpointsEnabled,
- alignmentTimeout,
+ alignedCheckpointTimeout,
checkpointIdOfIgnoredInFlightData);
}
@@ -325,9 +329,9 @@ public class CheckpointCoordinatorConfiguration implements Serializable {
return this;
}
- public CheckpointCoordinatorConfigurationBuilder setAlignmentTimeout(
- long alignmentTimeout) {
- this.alignmentTimeout = alignmentTimeout;
+ public CheckpointCoordinatorConfigurationBuilder setAlignedCheckpointTimeout(
+ long alignedCheckpointTimeout) {
+ this.alignedCheckpointTimeout = alignedCheckpointTimeout;
return this;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java
index dc02c7f..8ba71d9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointConfigHandler.java
@@ -131,7 +131,7 @@ public class CheckpointConfigHandler
checkpointStorageName,
checkpointCoordinatorConfiguration.isUnalignedCheckpointsEnabled(),
checkpointCoordinatorConfiguration.getTolerableCheckpointFailureNumber(),
- checkpointCoordinatorConfiguration.getAlignmentTimeout());
+ checkpointCoordinatorConfiguration.getAlignedCheckpointTimeout());
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index 22de6e2..4708598 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -623,7 +623,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
.setExecutionGraph(graph)
.setCheckpointCoordinatorConfiguration(
CheckpointCoordinatorConfiguration.builder()
- .setAlignmentTimeout(Long.MAX_VALUE)
+ .setAlignedCheckpointTimeout(Long.MAX_VALUE)
.setMaxConcurrentCheckpoints(Integer.MAX_VALUE)
.build())
.setTimer(manuallyTriggeredScheduledExecutor)
@@ -3440,7 +3440,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
.setExecutionGraph(graph)
.setCheckpointCoordinatorConfiguration(
CheckpointCoordinatorConfiguration.builder()
- .setAlignmentTimeout(Long.MAX_VALUE)
+ .setAlignedCheckpointTimeout(Long.MAX_VALUE)
.setMaxConcurrentCheckpoints(Integer.MAX_VALUE)
.build())
.setTimer(manuallyTriggeredScheduledExecutor)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointOptionsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointOptionsTest.java
index a55fb2e..156d187 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointOptionsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointOptionsTest.java
@@ -26,7 +26,7 @@ import org.junit.Test;
import java.util.Random;
-import static org.apache.flink.runtime.checkpoint.CheckpointOptions.NO_ALIGNMENT_TIME_OUT;
+import static org.apache.flink.runtime.checkpoint.CheckpointOptions.NO_ALIGNED_CHECKPOINT_TIME_OUT;
import static org.apache.flink.runtime.checkpoint.CheckpointType.CHECKPOINT;
import static org.apache.flink.runtime.checkpoint.CheckpointType.SAVEPOINT;
import static org.junit.Assert.assertArrayEquals;
@@ -83,25 +83,28 @@ public class CheckpointOptionsTest {
CHECKPOINT,
location,
AlignmentType.UNALIGNED,
- NO_ALIGNMENT_TIME_OUT)
+ NO_ALIGNED_CHECKPOINT_TIME_OUT)
.needsAlignment());
assertTrue(
new CheckpointOptions(
- CHECKPOINT, location, AlignmentType.ALIGNED, NO_ALIGNMENT_TIME_OUT)
+ CHECKPOINT,
+ location,
+ AlignmentType.ALIGNED,
+ NO_ALIGNED_CHECKPOINT_TIME_OUT)
.needsAlignment());
assertTrue(
new CheckpointOptions(
CHECKPOINT,
location,
AlignmentType.FORCED_ALIGNED,
- NO_ALIGNMENT_TIME_OUT)
+ NO_ALIGNED_CHECKPOINT_TIME_OUT)
.needsAlignment());
assertFalse(
new CheckpointOptions(
CHECKPOINT,
location,
AlignmentType.AT_LEAST_ONCE,
- NO_ALIGNMENT_TIME_OUT)
+ NO_ALIGNED_CHECKPOINT_TIME_OUT)
.needsAlignment());
}
@@ -111,7 +114,7 @@ public class CheckpointOptionsTest {
CheckpointStorageLocationReference.getDefault();
assertTimeoutable(CheckpointOptions.alignedWithTimeout(location, 10), false, true, 10);
assertTimeoutable(
- CheckpointOptions.unaligned(location), true, false, NO_ALIGNMENT_TIME_OUT);
+ CheckpointOptions.unaligned(location), true, false, NO_ALIGNED_CHECKPOINT_TIME_OUT);
assertTimeoutable(
CheckpointOptions.alignedWithTimeout(location, 10).withUnalignedUnsupported(),
false,
@@ -121,7 +124,7 @@ public class CheckpointOptionsTest {
CheckpointOptions.unaligned(location).withUnalignedUnsupported(),
false,
false,
- NO_ALIGNMENT_TIME_OUT);
+ NO_ALIGNED_CHECKPOINT_TIME_OUT);
}
@Test
@@ -160,6 +163,6 @@ public class CheckpointOptionsTest {
assertEquals("need alignment", !isUnaligned, options.needsAlignment());
assertEquals("unaligned", isUnaligned, options.isUnalignedCheckpoint());
assertEquals("timeoutable", isTimeoutable, options.isTimeoutable());
- assertEquals("timeout", timeout, options.getAlignmentTimeout());
+ assertEquals("timeout", timeout, options.getAlignedCheckpointTimeout());
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
index a087a39..7efd3b4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
@@ -98,8 +98,9 @@ public class CheckpointConfig implements java.io.Serializable {
private long checkpointIdOfIgnoredInFlightData =
DEFAULT_CHECKPOINT_ID_OF_IGNORED_IN_FLIGHT_DATA;
- private Duration alignmentTimeout =
- ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT.defaultValue();
+ /** The delay from the start of checkpoint after which AC switches to UC. */
+ private Duration alignedCheckpointTimeout =
+ ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT.defaultValue();
/** Flag to enable approximate local recovery. */
private boolean approximateLocalRecovery;
@@ -149,7 +150,7 @@ public class CheckpointConfig implements java.io.Serializable {
this.preferCheckpointForRecovery = checkpointConfig.preferCheckpointForRecovery;
this.tolerableCheckpointFailureNumber = checkpointConfig.tolerableCheckpointFailureNumber;
this.unalignedCheckpointsEnabled = checkpointConfig.isUnalignedCheckpointsEnabled();
- this.alignmentTimeout = checkpointConfig.alignmentTimeout;
+ this.alignedCheckpointTimeout = checkpointConfig.alignedCheckpointTimeout;
this.approximateLocalRecovery = checkpointConfig.isApproximateLocalRecoveryEnabled();
this.externalizedCheckpointCleanup = checkpointConfig.externalizedCheckpointCleanup;
this.forceCheckpointing = checkpointConfig.forceCheckpointing;
@@ -540,26 +541,57 @@ public class CheckpointConfig implements java.io.Serializable {
/**
* Only relevant if {@link #unalignedCheckpointsEnabled} is enabled.
*
- * <p>If {@link #alignmentTimeout} has value equal to <code>0</code>, checkpoints will always
- * start unaligned.
+ * <p>If {@link #alignedCheckpointTimeout} has value equal to <code>0</code>, checkpoints will
+ * always start unaligned.
*
- * <p>If {@link #alignmentTimeout} has value greater then <code>0</code>, checkpoints will start
- * aligned. If during checkpointing, checkpoint start delay exceeds this {@link
- * #alignmentTimeout}, alignment will timeout and checkpoint will start working as unaligned
- * checkpoint.
+ * <p>If {@link #alignedCheckpointTimeout} has value greater then <code>0</code>, checkpoints
+ * will start aligned. If during checkpointing, checkpoint start delay exceeds this {@link
+ * #alignedCheckpointTimeout}, alignment will timeout and checkpoint will start working as
+ * unaligned checkpoint.
+ *
+ * @deprecated Use {@link #setAlignedCheckpointTimeout(Duration)} instead.
*/
+ @Deprecated
@PublicEvolving
public void setAlignmentTimeout(Duration alignmentTimeout) {
- this.alignmentTimeout = alignmentTimeout;
+ this.alignedCheckpointTimeout = alignmentTimeout;
}
/**
* @return value of alignment timeout, as configured via {@link #setAlignmentTimeout(Duration)}
* or {@link ExecutionCheckpointingOptions#ALIGNMENT_TIMEOUT}.
+ * @deprecated User {@link #getAlignedCheckpointTimeout()} instead.
*/
+ @Deprecated
@PublicEvolving
public Duration getAlignmentTimeout() {
- return alignmentTimeout;
+ return alignedCheckpointTimeout;
+ }
+
+ /**
+ * @return value of alignment timeout, as configured via {@link
+ * #setAlignedCheckpointTimeout(Duration)} or {@link
+ * ExecutionCheckpointingOptions#ALIGNED_CHECKPOINT_TIMEOUT}.
+ */
+ @PublicEvolving
+ public Duration getAlignedCheckpointTimeout() {
+ return alignedCheckpointTimeout;
+ }
+
+ /**
+ * Only relevant if {@link #unalignedCheckpointsEnabled} is enabled.
+ *
+ * <p>If {@link #alignedCheckpointTimeout} has value equal to <code>0</code>, checkpoints will
+ * always start unaligned.
+ *
+ * <p>If {@link #alignedCheckpointTimeout} has value greater then <code>0</code>, checkpoints
+ * will start aligned. If during checkpointing, checkpoint start delay exceeds this {@link
+ * #alignedCheckpointTimeout}, alignment will timeout and checkpoint will start working as
+ * unaligned checkpoint.
+ */
+ @PublicEvolving
+ public void setAlignedCheckpointTimeout(Duration alignedCheckpointTimeout) {
+ this.alignedCheckpointTimeout = alignedCheckpointTimeout;
}
/**
@@ -784,8 +816,8 @@ public class CheckpointConfig implements java.io.Serializable {
.getOptional(ExecutionCheckpointingOptions.CHECKPOINT_ID_OF_IGNORED_IN_FLIGHT_DATA)
.ifPresent(this::setCheckpointIdOfIgnoredInFlightData);
configuration
- .getOptional(ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT)
- .ifPresent(this::setAlignmentTimeout);
+ .getOptional(ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT)
+ .ifPresent(this::setAlignedCheckpointTimeout);
configuration
.getOptional(ExecutionCheckpointingOptions.FORCE_UNALIGNED)
.ifPresent(this::setForceUnalignedCheckpoints);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java
index 1c9d97a..c8cefe7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/ExecutionCheckpointingOptions.java
@@ -166,6 +166,30 @@ public class ExecutionCheckpointingOptions {
TextElement.code(MAX_CONCURRENT_CHECKPOINTS.key()))
.build());
+ public static final ConfigOption<Duration> ALIGNED_CHECKPOINT_TIMEOUT =
+ ConfigOptions.key("execution.checkpointing.aligned-checkpoint-timeout")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(0L))
+ .withDeprecatedKeys("execution.checkpointing.alignment-timeout")
+ .withDescription(
+ Description.builder()
+ .text(
+ "Only relevant if %s is enabled.",
+ TextElement.code(ENABLE_UNALIGNED.key()))
+ .linebreak()
+ .linebreak()
+ .text(
+ "If timeout is 0, checkpoints will always start unaligned.")
+ .linebreak()
+ .linebreak()
+ .text(
+ "If timeout has a positive value, checkpoints will start aligned. "
+ + "If during checkpointing, checkpoint start delay exceeds this timeout, alignment "
+ + "will timeout and checkpoint barrier will start working as unaligned checkpoint.")
+ .build());
+
+ /** @deprecated Use {@link #ALIGNED_CHECKPOINT_TIMEOUT} instead. */
+ @Deprecated
public static final ConfigOption<Duration> ALIGNMENT_TIMEOUT =
ConfigOptions.key("execution.checkpointing.alignment-timeout")
.durationType()
@@ -173,7 +197,8 @@ public class ExecutionCheckpointingOptions {
.withDescription(
Description.builder()
.text(
- "Only relevant if %s is enabled.",
+ "Deprecated. %s should be used instead. Only relevant if %s is enabled.",
+ TextElement.code(ALIGNED_CHECKPOINT_TIMEOUT.key()),
TextElement.code(ENABLE_UNALIGNED.key()))
.linebreak()
.linebreak()
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
index 085a327..6f9e674 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
@@ -462,12 +462,13 @@ public class StreamConfig implements Serializable {
return getCheckpointMode() == CheckpointingMode.EXACTLY_ONCE;
}
- public Duration getAlignmentTimeout() {
- return config.get(ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT);
+ public Duration getAlignedCheckpointTimeout() {
+ return config.get(ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT);
}
- public void setAlignmentTimeout(Duration alignmentTimeout) {
- config.set(ExecutionCheckpointingOptions.ALIGNMENT_TIMEOUT, alignmentTimeout);
+ public void setAlignedCheckpointTimeout(Duration alignedCheckpointTimeout) {
+ config.set(
+ ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, alignedCheckpointTimeout);
}
public void setOutEdgesInOrder(List<StreamEdge> outEdgeList) {
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index f06aa05..6e96655 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -728,7 +728,7 @@ public class StreamingJobGraphGenerator {
config.setCheckpointingEnabled(checkpointCfg.isCheckpointingEnabled());
config.setCheckpointMode(getCheckpointingMode(checkpointCfg));
config.setUnalignedCheckpointsEnabled(checkpointCfg.isUnalignedCheckpointsEnabled());
- config.setAlignmentTimeout(checkpointCfg.getAlignmentTimeout());
+ config.setAlignedCheckpointTimeout(checkpointCfg.getAlignedCheckpointTimeout());
for (int i = 0; i < vertex.getStatePartitioners().length; i++) {
config.setStatePartitioner(i, vertex.getStatePartitioners()[i]);
@@ -1315,7 +1315,8 @@ public class StreamingJobGraphGenerator {
.setUnalignedCheckpointsEnabled(cfg.isUnalignedCheckpointsEnabled())
.setCheckpointIdOfIgnoredInFlightData(
cfg.getCheckpointIdOfIgnoredInFlightData())
- .setAlignmentTimeout(cfg.getAlignmentTimeout().toMillis())
+ .setAlignedCheckpointTimeout(
+ cfg.getAlignedCheckpointTimeout().toMillis())
.build(),
serializedStateBackend,
streamGraph.isChangelogStateBackendEnabled(),
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
index fcca333..e2ad1cb 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
@@ -271,7 +271,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
private void registerAlignmentTimer(CheckpointBarrier announcedBarrier) {
long alignedCheckpointTimeout =
- announcedBarrier.getCheckpointOptions().getAlignmentTimeout();
+ announcedBarrier.getCheckpointOptions().getAlignedCheckpointTimeout();
long timePassedSinceCheckpointStart =
getClock().absoluteTimeMillis() - announcedBarrier.getTimestamp();
@@ -436,7 +436,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
public boolean isTimedOut(CheckpointBarrier barrier) {
return barrier.getCheckpointOptions().isTimeoutable()
&& barrier.getId() <= currentCheckpointId
- && barrier.getCheckpointOptions().getAlignmentTimeout()
+ && barrier.getCheckpointOptions().getAlignedCheckpointTimeout()
< (getClock().absoluteTimeMillis() - barrier.getTimestamp());
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
index b77710e..5ba8321 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
@@ -148,7 +148,7 @@ public class SourceOperatorStreamTask<T> extends StreamTask<T, SourceOperator<T,
CheckpointStorageLocationReference.getDefault(),
configuration.isExactlyOnceCheckpointMode(),
configuration.isUnalignedCheckpointsEnabled(),
- configuration.getAlignmentTimeout().toMillis());
+ configuration.getAlignedCheckpointTimeout().toMillis());
final long timestamp = System.currentTimeMillis();
final CheckpointMetaData checkpointMetaData =
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
index ad5b99f..1d132e1 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -109,7 +109,7 @@ public class SourceStreamTask<
CheckpointStorageLocationReference.getDefault(),
configuration.isExactlyOnceCheckpointMode(),
configuration.isUnalignedCheckpointsEnabled(),
- configuration.getAlignmentTimeout().toMillis());
+ configuration.getAlignedCheckpointTimeout().toMillis());
final long timestamp = System.currentTimeMillis();
final CheckpointMetaData checkpointMetaData =
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
index c541e67..da70c99 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
@@ -300,7 +300,7 @@ public class MultipleInputStreamTaskChainedSourcesCheckpointingTest {
CheckpointStorageLocationReference.getDefault(),
config.isExactlyOnceCheckpointMode(),
config.isUnalignedCheckpointsEnabled(),
- config.getAlignmentTimeout().toMillis());
+ config.getAlignedCheckpointTimeout().toMillis());
return new CheckpointBarrier(
metaData.getCheckpointId(), metaData.getTimestamp(), checkpointOptions);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
index 6f5e8eb..7a0669d 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
@@ -1046,7 +1046,7 @@ public class MultipleInputStreamTaskTest {
MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
.modifyExecutionConfig(config -> config.enableObjectReuse())
.modifyStreamConfig(config -> config.setUnalignedCheckpointsEnabled(unaligned))
- .modifyStreamConfig(config -> config.setAlignmentTimeout(Duration.ZERO))
+ .modifyStreamConfig(config -> config.setAlignedCheckpointTimeout(Duration.ZERO))
.addInput(BasicTypeInfo.STRING_TYPE_INFO)
.addSourceInput(
new SourceOperatorFactory<>(