You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/07/23 11:52:42 UTC
[GitHub] [flink] Myasuka opened a new pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Myasuka opened a new pull request #16582:
URL: https://github.com/apache/flink/pull/16582
## What is the purpose of the change
Introduce notification of subsumed checkpoint
## Brief change log
Introduce notification of subsumed checkpoint. Add a new `InternalCheckpointListener` and related implementations.
## Verifying this change
This change added tests and can be verified as follows:
SubtaskCheckpointCoordinatorTest
CheckpointCoordinatorTest
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677217464
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
> Actually, current CheckpointStore#addCheckpoint has the meaning that this is an atomic operation which will add the completed checkpoint and subsumed an older checkpoint. I think return the subsumed checkpoint is clearly for this operation.
To me, it's only clear that it might subsume some older checkpoints. It is unclear whether returned is the first subsumed, the latest subsumed, or the added one.
> If we adopt the API long getLatestSubsumedCheckpointID()`, we will miss some information that when the checkpoint is subsumed and we might also need to check whether the latest subsumed checkpoint ID has been changed.
I think there is no requirement for the CheckpointCoordinator to report strictly latest ID (it will be outdated right after the RPC). It must only be non-decreasing. If it somehow becomes greater than the added one we could use the smallest from them.
And currently Store is used from the same thread, so calling `addCheckpoint` and then `getLatestSubsumed` is safe (I don't think this concurrency model should be changed).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677174708
##########
File path: flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.java
##########
@@ -0,0 +1,52 @@
+/*
+
+* 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.
+
+*/
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * The internal checkpoint listener add another {@link #notifyCheckpointSubsumed(long)} interface
+ * for internal usage.
+ */
+@Internal
+public interface InternalCheckpointListener extends CheckpointListener {
+
+ /**
+ * A invalid checkpoint id, which could be used in {@link #notifyCheckpointSubsumed(long)} to
+ * represent no checkpoint needs to be subsumed.
+ */
+ long INVALID_CHECKPOINT_ID = -1;
+
+ /**
+ * This method is called as a notification once a distributed checkpoint has been subsumed.
+ *
+ * <p>These notifications are "best effort", meaning they can sometimes be skipped.
+ *
+ * <p>This method is very rarely necessary to implement. The "best effort" guarantee, together
+ * with the fact that this method should not result in discarding any data (per the "Checkpoint
+ * Subsuming Contract") means it is mainly useful for earlier cleanups of auxiliary resources.
+ *
+ * @param checkpointId The ID of the checkpoint that has been subsumed.
+ * @throws Exception This method can propagate exceptions, which leads to a failure/recovery for
+ * the task or job.
+ */
+ default void notifyCheckpointSubsumed(long checkpointId) throws Exception {}
Review comment:
The original idea is to like current `CheckpointListener#notifyCheckpointAborted` which makes a default implementation to not let the normal users know.
After making the interface not a default no-op implementation, I tried to make those abstract class implement this necessary interface to make the changes as least as possible. And it seems okay for those changes.
I will update in the following commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r751027356
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1405,59 +1406,80 @@ private void declineCheckpoint(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final TaskInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING) {
- checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
- try {
- ((CheckpointableTask) invokable).notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.COMPLETE);
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final TaskInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID, latestCompletedCheckpointId, NotifyCheckpointOperation.ABORT);
+ }
- if (executionState == ExecutionState.RUNNING) {
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.SUBSUME);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ long latestCompletedCheckpointId,
+ NotifyCheckpointOperation notifyCheckpointOperation) {
+ TaskInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
try {
- ((CheckpointableTask) invokable)
- .notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointAbortAsync(
+ checkpointId, latestCompletedCheckpointId);
+ break;
+ case COMPLETE:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointCompleteAsync(checkpointId);
+ break;
+ case SUBSUME:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointSubsumedAsync(checkpointId);
+ }
} catch (RejectedExecutionException ex) {
// This may happen if the mailbox is closed. It means that the task is shutting
// down, so we just ignore it.
LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
+ "Notify checkpoint {}} {} for {} ({}) was rejected by the mailbox.",
+ notifyCheckpointOperation,
+ checkpointId,
taskNameWithSubtask,
executionId);
} catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ case COMPLETE:
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ failExternally(
+ new RuntimeException(
+ String.format(
+ "Error while notify checkpoint %s.",
+ notifyCheckpointOperation),
+ t));
+ }
+ break;
+ case SUBSUME:
+ // just rethrow the throwable out as we do not expect notification of
+ // subsume could fail the task.
+ ExceptionUtils.rethrow(t);
Review comment:
From my point of view, we need to ensure task failover directly on the case of checkpoint complete notification to avoid unexpected data lost. However, I don't think we should also apply this rule on checkpoint subsumed notification as there is no strict data completeness request here. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8ca91f13df7b3d17d2137e008e51acbcb4067c97 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991)
* 1168b48d412c2ac15da72d06bcdff90c2120f2c8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1168b48d412c2ac15da72d06bcdff90c2120f2c8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725)
* 4575a58b2365b9ab070606108ef6ba1b4507dc24 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677260088
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
##########
@@ -382,6 +382,11 @@ public void notifyCheckpointAborted(long checkpointId) throws Exception {
stateHandler.notifyCheckpointAborted(checkpointId);
}
+ @Override
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ stateHandler.notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
I see, thanks for your reminder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 36cfdc4abcad6028059c134f323e963d455921e8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078)
* dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r751022321
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -46,18 +46,24 @@
class CheckpointSubsumeHelper {
private static final Logger LOG = LoggerFactory.getLogger(CheckpointSubsumeHelper.class);
- public static void subsume(
+ public static Optional<CompletedCheckpoint> subsume(
Deque<CompletedCheckpoint> checkpoints, int numRetain, SubsumeAction subsumeAction)
throws Exception {
if (checkpoints.isEmpty() || checkpoints.size() <= numRetain) {
- return;
+ return Optional.empty();
}
CompletedCheckpoint latest = checkpoints.peekLast();
+ CompletedCheckpoint lastSubsumedCheckpoint = null;
Review comment:
Thanks for the suggestion, I will refactor it in the next upadte.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r742598674
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -46,18 +48,24 @@
class CheckpointSubsumeHelper {
private static final Logger LOG = LoggerFactory.getLogger(CheckpointSubsumeHelper.class);
- public static void subsume(
+ @Nullable
+ public static CompletedCheckpoint subsume(
Deque<CompletedCheckpoint> checkpoints, int numRetain, SubsumeAction subsumeAction)
throws Exception {
if (checkpoints.isEmpty() || checkpoints.size() <= numRetain) {
- return;
+ return null;
Review comment:
I see, I'll refactor API to use `Optional`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 935abc391163053cdc2d5fb35d5f321844f0e8b1 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 935abc391163053cdc2d5fb35d5f321844f0e8b1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1168b48d412c2ac15da72d06bcdff90c2120f2c8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r742651490
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1352,84 +1352,73 @@ public void triggerCheckpointBarrier(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final AbstractInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING && invokable != null) {
- try {
- invokable.notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) -> invokable.notifyCheckpointCompleteAsync(checkpointID),
+ "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
+ "Error while confirming checkpoint {}.",
+ (t) -> failExternally(new RuntimeException("Error while confirming checkpoint", t)),
+ "Ignoring checkpoint commit notification for non-running task {}.");
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final AbstractInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) ->
+ invokable.notifyCheckpointAbortAsync(
+ checkpointID, latestCompletedCheckpointId),
+ "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
+ "Error while aborting checkpoint {}.",
+ (t) -> failExternally(new RuntimeException("Error while aborting checkpoint", t)),
+ "Ignoring checkpoint aborted notification for non-running task {}.");
+ }
- if (executionState == ExecutionState.RUNNING && invokable != null) {
- try {
- invokable.notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
- }
- }
- } else {
- LOG.info(
- "Ignoring checkpoint aborted notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) -> invokable.notifyCheckpointSubsumedAsync(checkpointID),
+ "Notify checkpoint subsume {} for {} ({}) was rejected by the mailbox",
+ "Error while subsuming checkpoint {}.",
+ null,
+ "Ignoring checkpoint subsume notification for non-running task {}.");
}
- public void notifyCheckpointSubsumed(long checkpointId) {
+ private void notifyCheckpoint(
+ long checkpointId,
+ Consumer<AbstractInvokable> invokableNotifier,
+ String logFormatIfRejected,
+ String logFormatIfFailedToNotify,
+ @Nullable Consumer<Throwable> handleThrowableIfTaskRunning,
+ String logFormatIfNotRunning) {
Review comment:
Thanks for the suggestion of introducing `enum` of `NotifyCheckpointOperation`, I think this could help reduce code duplication and also make the code more readable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677193306
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
Actually, current `CheckpointStore#addCheckpoint` has the meaning that this is an atomic operation which will add the completed checkpoint and subsumed an older checkpoint.` I think return the subsumed checkpoint is clearly for this operation.
If we adopt the API `long getLatestSubsumedCheckpointID()`, we will miss some information that when the checkpoint is subsumed and we might also need to check whether the latest subsumed checkpoint ID has been changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474)
* 8ca91f13df7b3d17d2137e008e51acbcb4067c97 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r709949196
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
edit: actually after removing field `@Nullable private CompletedCheckpoint lastSubsumedCheckpoint;` I don't see actually much difference between those two approaches. Is there a problem that I'm missing?
Note it doesn't help that the changes are not squashed, ant that the counter proposal is also a separate commit on top of this PR. It makes reviewing and comparing both options very difficult.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 935abc391163053cdc2d5fb35d5f321844f0e8b1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886)
* 10466b2d0065049cc8d003cdfc1395785d279e23 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677268152
##########
File path: flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.java
##########
@@ -0,0 +1,52 @@
+/*
+
+* 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.
+
+*/
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * The internal checkpoint listener add another {@link #notifyCheckpointSubsumed(long)} interface
+ * for internal usage.
+ */
+@Internal
+public interface InternalCheckpointListener extends CheckpointListener {
Review comment:
My original idea is to add the new method in `CheckpointListener`. However, current `CheckpointListener` is already annotated as `public` and the semantics of a checkpoint subsumed should be acted as an internal operation only for runtime classes.
Maybe we can see the relationship as `InternalKvState` and `State`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677214452
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManager.java
##########
@@ -42,7 +42,7 @@
* <p>This interface also offers the complementary method that provides access to previously saved
* state of operator instances in the task for restore purposes.
*/
-public interface TaskStateManager extends CheckpointListener, AutoCloseable {
+public interface TaskStateManager extends InternalCheckpointListener, AutoCloseable {
Review comment:
I previously thought that `TaskStateManagerImpl` could also leverage the `notifyCheckpointSubsumed` API to help remove local states. This is actually another story and not seems so valuable in current implementation.
I will remove this in next commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
}, {
"hash" : "ab089745cc9841edb718c33a8ea3224775e38766",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862",
"triggerID" : "ab089745cc9841edb718c33a8ea3224775e38766",
"triggerType" : "PUSH"
}, {
"hash" : "f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ab089745cc9841edb718c33a8ea3224775e38766 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862)
* f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
}, {
"hash" : "ab089745cc9841edb718c33a8ea3224775e38766",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862",
"triggerID" : "ab089745cc9841edb718c33a8ea3224775e38766",
"triggerType" : "PUSH"
}, {
"hash" : "f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27900",
"triggerID" : "f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ab089745cc9841edb718c33a8ea3224775e38766 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862)
* f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27900)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
}, {
"hash" : "ab089745cc9841edb718c33a8ea3224775e38766",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862",
"triggerID" : "ab089745cc9841edb718c33a8ea3224775e38766",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ab089745cc9841edb718c33a8ea3224775e38766 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410)
* 8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r709876746
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -46,18 +48,24 @@
class CheckpointSubsumeHelper {
private static final Logger LOG = LoggerFactory.getLogger(CheckpointSubsumeHelper.class);
- public static void subsume(
+ @Nullable
+ public static CompletedCheckpoint subsume(
Deque<CompletedCheckpoint> checkpoints, int numRetain, SubsumeAction subsumeAction)
throws Exception {
if (checkpoints.isEmpty() || checkpoints.size() <= numRetain) {
- return;
+ return null;
Review comment:
`Optional` for compile time checks?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
##########
@@ -339,78 +343,107 @@ public void checkpointState(
public void notifyCheckpointComplete(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
- Exception previousException = null;
- if (!isRunning.get()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for not-running task {}",
- checkpointId,
- taskName);
- } else if (operatorChain.isFinishedOnRestore()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for finished on restore task {}",
- checkpointId,
- taskName);
- } else {
- LOG.debug(
- "Notification of completed checkpoint {} for task {}", checkpointId, taskName);
- for (StreamOperatorWrapper<?, ?> operatorWrapper :
- operatorChain.getAllOperators(true)) {
- try {
- operatorWrapper.notifyCheckpointComplete(checkpointId);
- } catch (Exception e) {
- previousException = ExceptionUtils.firstOrSuppressed(e, previousException);
- }
- }
- }
- env.getTaskStateManager().notifyCheckpointComplete(checkpointId);
- ExceptionUtils.tryRethrowException(previousException);
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of complete checkpoint {} for not-running task {}",
+ "Ignoring notification of complete checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointComplete(checkpointId),
+ (tsm) -> tsm.notifyCheckpointComplete(checkpointId));
}
@Override
public void notifyCheckpointAborted(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of aborted checkpoint {} for not-running task {}",
+ "Ignoring notification of aborted checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ () -> {
+ boolean canceled = cancelAsyncCheckpointRunnable(checkpointId);
+
+ if (!canceled) {
+ if (checkpointId > lastCheckpointId) {
+ // only record checkpoints that have not triggered on task side.
+ abortedCheckpointIds.add(checkpointId);
+ }
+ }
+
+ channelStateWriter.abort(
+ checkpointId,
+ new CancellationException("checkpoint aborted via notification"),
+ false);
+ },
+ (opw) -> opw.getStreamOperator().notifyCheckpointAborted(checkpointId),
+ (tsm) -> tsm.notifyCheckpointAborted(checkpointId));
+ }
+
+ @Override
+ public void notifyCheckpointSubsumed(
+ long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
+ throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of subsumed checkpoint {} for not-running task {}",
+ "Ignoring notification of subsumed checkpoint {} for finished on restore task {}",
+ "Notification of subsumed checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointSubsumed(checkpointId),
+ null);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ OperatorChain<?, ?> operatorChain,
+ Supplier<Boolean> isRunning,
+ String logFormatIfNotRunning,
+ String logFormatIfFinishedOnRestore,
+ String logFormatIfNotificationStart,
+ @Nullable Runnable runnableBeforeNotifyOperator,
+ @Nonnull ThrowingConsumer<StreamOperatorWrapper<?, ?>, Exception> operatorConsumer,
+ @Nullable ThrowingConsumer<TaskStateManager, Exception> taskStateManagerConsumer)
+ throws Exception {
Review comment:
Instead of using `Runnable` and `ThrowingConsumer` in order to deduplicate the code, I would suggest to consider deduplicating it by decomposing this `notifyCheckpoint` method into smaller parts. For example:
```
public void notifyCheckpointAborted(...) {
if (notifyCheckpointOperationrShouldBeExecuted()) {
// inlined content of the `runnableBeforeNotifyOperator`
operatorChain.forEach((operatorWrapper) -> operatorWrapper.notifyCheckpointSubsumed(checkpointId));
env.getTaskStateManager().notifyCheckpointAborted(checkpointId);
}
}
public void notifyCheckpointSubsumed(...) {
if (notifyCheckpointOperationrShouldBeExecuted()) {
operatorChain.forEach((operatorWrapper) -> operatorWrapper.notifyCheckpointAborted(checkpointId));
}
}
private boolean notifyCheckpointOperationrShouldBeExecuted(...) {
// isRunning/isFinishedOnRestore checks + logging
}
```
Yes, we would still have a bit of code duplication, but it would avoid the problem of generic `Runnable` `ThrowingConsumer` that are mangling/hiding what the code is doing. In your current version to understand what this code is doing, you have to keep jumping back and forth between `notifyCheckpointAborted()` method (where those `Runnable` and `ThrowingConsumers` are defined) and the actual `notifyCheckpoint` implementation. What makes it even worse, is that you can not use "find implementations" from the IDE to help you, as there are just so many `Runnable`s in the project. So maybe a bit of code duplication is not that bad in that case? (I'm not 100% sure)
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
##########
@@ -339,78 +343,107 @@ public void checkpointState(
public void notifyCheckpointComplete(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
- Exception previousException = null;
- if (!isRunning.get()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for not-running task {}",
- checkpointId,
- taskName);
- } else if (operatorChain.isFinishedOnRestore()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for finished on restore task {}",
- checkpointId,
- taskName);
- } else {
- LOG.debug(
- "Notification of completed checkpoint {} for task {}", checkpointId, taskName);
- for (StreamOperatorWrapper<?, ?> operatorWrapper :
- operatorChain.getAllOperators(true)) {
- try {
- operatorWrapper.notifyCheckpointComplete(checkpointId);
- } catch (Exception e) {
- previousException = ExceptionUtils.firstOrSuppressed(e, previousException);
- }
- }
- }
- env.getTaskStateManager().notifyCheckpointComplete(checkpointId);
- ExceptionUtils.tryRethrowException(previousException);
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of complete checkpoint {} for not-running task {}",
+ "Ignoring notification of complete checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointComplete(checkpointId),
+ (tsm) -> tsm.notifyCheckpointComplete(checkpointId));
}
@Override
public void notifyCheckpointAborted(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of aborted checkpoint {} for not-running task {}",
+ "Ignoring notification of aborted checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ () -> {
+ boolean canceled = cancelAsyncCheckpointRunnable(checkpointId);
+
+ if (!canceled) {
+ if (checkpointId > lastCheckpointId) {
+ // only record checkpoints that have not triggered on task side.
+ abortedCheckpointIds.add(checkpointId);
+ }
+ }
+
+ channelStateWriter.abort(
+ checkpointId,
+ new CancellationException("checkpoint aborted via notification"),
+ false);
+ },
+ (opw) -> opw.getStreamOperator().notifyCheckpointAborted(checkpointId),
+ (tsm) -> tsm.notifyCheckpointAborted(checkpointId));
Review comment:
please no abbreviations. By Looking at this code I have no idea what `opw` or `tsm` are.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
> In general, we should make the minimal changes to existing interfaces.
I think changes in both of those proposals are quite comparable. Adding one getter vs changing return type of a method to substitute for this getter.
> Adding method of CompletedCheckpointStore#getLatestSubsumedCheckpointID would introduce another method and would also make us lost the timing when the checkpoint is subsumed.
What is the problem here? Could you elaborate what are you worried about?
On the other hand, I like the @rkhachatryan's approach that there seems to be clear owner of this knowledge, what was/is the last subsumed checkpoint and as a result of this, we don't need to modify `CheckpointCoordinator` that much. The approach with returning this information in the `addCheckpoint` method, that needs to be stored/cached externally in the `CheckpointCoordinator` introduces more dependencies in the code.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1352,84 +1352,73 @@ public void triggerCheckpointBarrier(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final AbstractInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING && invokable != null) {
- try {
- invokable.notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) -> invokable.notifyCheckpointCompleteAsync(checkpointID),
+ "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
+ "Error while confirming checkpoint {}.",
+ (t) -> failExternally(new RuntimeException("Error while confirming checkpoint", t)),
+ "Ignoring checkpoint commit notification for non-running task {}.");
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final AbstractInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) ->
+ invokable.notifyCheckpointAbortAsync(
+ checkpointID, latestCompletedCheckpointId),
+ "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
+ "Error while aborting checkpoint {}.",
+ (t) -> failExternally(new RuntimeException("Error while aborting checkpoint", t)),
+ "Ignoring checkpoint aborted notification for non-running task {}.");
+ }
- if (executionState == ExecutionState.RUNNING && invokable != null) {
- try {
- invokable.notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
- }
- }
- } else {
- LOG.info(
- "Ignoring checkpoint aborted notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) -> invokable.notifyCheckpointSubsumedAsync(checkpointID),
+ "Notify checkpoint subsume {} for {} ({}) was rejected by the mailbox",
+ "Error while subsuming checkpoint {}.",
+ null,
+ "Ignoring checkpoint subsume notification for non-running task {}.");
}
- public void notifyCheckpointSubsumed(long checkpointId) {
+ private void notifyCheckpoint(
+ long checkpointId,
+ Consumer<AbstractInvokable> invokableNotifier,
+ String logFormatIfRejected,
+ String logFormatIfFailedToNotify,
+ @Nullable Consumer<Throwable> handleThrowableIfTaskRunning,
+ String logFormatIfNotRunning) {
Review comment:
For me this method has the same problem as in my previous comment with generic Consumer's. However here I would propose to add an `enum NotifyCheckpointOperation` with three values `ABORT`, `COMPLETE`, `SUBSUME`, and just switch/case it to replace `Consumer<AbstractInvokable> invokableNotifier`.
On the other hand `handleThrowableIfTaskRunning` could be replaced by boolean flag `boolean shouldFailExternallyOnException`, so the signature of this method could be:
```
private void notifyCheckpoint(
long checkpointId,
NotifyCheckpointOperation notifyCheckpointOperation,
boolean shouldFailExternallyOnException)
```
Thirdly, the log/error formats I think could be handled also easier, by just using something like:
```
String.format("Notify checkpoint {} {} for {} ({}) was rejected by the mailbox", notifyCheckpointOperation.toString(), ...);
```
instead of
```
String.format("Notify checkpoint abort {} for {} ({}) was rejected by the mailbox", ...);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5817be659ecb4b6124f01415e552de29f538960b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030)
* 36cfdc4abcad6028059c134f323e963d455921e8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
}, {
"hash" : "ab089745cc9841edb718c33a8ea3224775e38766",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862",
"triggerID" : "ab089745cc9841edb718c33a8ea3224775e38766",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4575a58b2365b9ab070606108ef6ba1b4507dc24 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764)
* ab089745cc9841edb718c33a8ea3224775e38766 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r743366210
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
##########
@@ -339,78 +343,107 @@ public void checkpointState(
public void notifyCheckpointComplete(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
- Exception previousException = null;
- if (!isRunning.get()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for not-running task {}",
- checkpointId,
- taskName);
- } else if (operatorChain.isFinishedOnRestore()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for finished on restore task {}",
- checkpointId,
- taskName);
- } else {
- LOG.debug(
- "Notification of completed checkpoint {} for task {}", checkpointId, taskName);
- for (StreamOperatorWrapper<?, ?> operatorWrapper :
- operatorChain.getAllOperators(true)) {
- try {
- operatorWrapper.notifyCheckpointComplete(checkpointId);
- } catch (Exception e) {
- previousException = ExceptionUtils.firstOrSuppressed(e, previousException);
- }
- }
- }
- env.getTaskStateManager().notifyCheckpointComplete(checkpointId);
- ExceptionUtils.tryRethrowException(previousException);
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of complete checkpoint {} for not-running task {}",
+ "Ignoring notification of complete checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointComplete(checkpointId),
+ (tsm) -> tsm.notifyCheckpointComplete(checkpointId));
}
@Override
public void notifyCheckpointAborted(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of aborted checkpoint {} for not-running task {}",
+ "Ignoring notification of aborted checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ () -> {
+ boolean canceled = cancelAsyncCheckpointRunnable(checkpointId);
+
+ if (!canceled) {
+ if (checkpointId > lastCheckpointId) {
+ // only record checkpoints that have not triggered on task side.
+ abortedCheckpointIds.add(checkpointId);
+ }
+ }
+
+ channelStateWriter.abort(
+ checkpointId,
+ new CancellationException("checkpoint aborted via notification"),
+ false);
+ },
+ (opw) -> opw.getStreamOperator().notifyCheckpointAborted(checkpointId),
+ (tsm) -> tsm.notifyCheckpointAborted(checkpointId));
+ }
+
+ @Override
+ public void notifyCheckpointSubsumed(
+ long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
+ throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of subsumed checkpoint {} for not-running task {}",
+ "Ignoring notification of subsumed checkpoint {} for finished on restore task {}",
+ "Notification of subsumed checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointSubsumed(checkpointId),
+ null);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ OperatorChain<?, ?> operatorChain,
+ Supplier<Boolean> isRunning,
+ String logFormatIfNotRunning,
+ String logFormatIfFinishedOnRestore,
+ String logFormatIfNotificationStart,
+ @Nullable Runnable runnableBeforeNotifyOperator,
+ @Nonnull ThrowingConsumer<StreamOperatorWrapper<?, ?>, Exception> operatorConsumer,
+ @Nullable ThrowingConsumer<TaskStateManager, Exception> taskStateManagerConsumer)
+ throws Exception {
Review comment:
The refactor also take the `enum` of `NotifyCheckpointOperation` as `Task` did.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677170632
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
The original idea is to not change the API, and I think a proper place is to change the return result of `CompletedCheckpointStore#addCheckpoint` from `void` to the subsumed `CompletedCheckpoint`. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r679131871
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
Thanks for Roman sharing the implementation of adding another method `CompletedCheckpointStore#getLatestSubsumedCheckpointID`. However, I still prefer to leverage previous `CompletedCheckpointStore#addCheckpoint` to know the last subsumed checkpoint for reasons:
1. In general, we should make the minimal changes to existing interfaces.
2. Previous method of `CompletedCheckpointStore#addCheckpoint` already act the role to subsume checkpoint/savepoint when adding a new complete checkpoint. We could achieve our goal If we just returned the subsumed checkpoint in previous method instead of return nothing.
3. Adding method of `CompletedCheckpointStore#getLatestSubsumedCheckpointID` would introduce another method and would also make us lost the timing when the checkpoint is subsumed.
After offline discussion with each other, we think maybe involving more guys could help this discussion more clear. Could you share some thoughts here? @StephanEwen @pnowojski
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r762973008
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1405,59 +1406,80 @@ private void declineCheckpoint(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final TaskInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING) {
- checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
- try {
- ((CheckpointableTask) invokable).notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.COMPLETE);
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final TaskInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID, latestCompletedCheckpointId, NotifyCheckpointOperation.ABORT);
+ }
- if (executionState == ExecutionState.RUNNING) {
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.SUBSUME);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ long latestCompletedCheckpointId,
+ NotifyCheckpointOperation notifyCheckpointOperation) {
+ TaskInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
try {
- ((CheckpointableTask) invokable)
- .notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointAbortAsync(
+ checkpointId, latestCompletedCheckpointId);
+ break;
+ case COMPLETE:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointCompleteAsync(checkpointId);
+ break;
+ case SUBSUME:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointSubsumedAsync(checkpointId);
+ }
} catch (RejectedExecutionException ex) {
// This may happen if the mailbox is closed. It means that the task is shutting
// down, so we just ignore it.
LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
+ "Notify checkpoint {}} {} for {} ({}) was rejected by the mailbox.",
+ notifyCheckpointOperation,
+ checkpointId,
taskNameWithSubtask,
executionId);
} catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ case COMPLETE:
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ failExternally(
+ new RuntimeException(
+ String.format(
+ "Error while notify checkpoint %s.",
+ notifyCheckpointOperation),
+ t));
+ }
+ break;
+ case SUBSUME:
+ // just rethrow the throwable out as we do not expect notification of
+ // subsume could fail the task.
+ ExceptionUtils.rethrow(t);
Review comment:
👍
So let's keep it as it is. (failure of `notifyCheckpointSubsumed` shouldn't cause the task to failover automatically).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8ca91f13df7b3d17d2137e008e51acbcb4067c97 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991)
* 1168b48d412c2ac15da72d06bcdff90c2120f2c8 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885587084
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 935abc391163053cdc2d5fb35d5f321844f0e8b1 (Fri Jul 23 11:55:54 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r680674905
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1399,6 +1399,33 @@ public void notifyCheckpointAborted(final long checkpointID) {
}
}
+ public void notifyCheckpointSubsumed(long checkpointId) {
+ AbstractInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
+
+ try {
+ invokable.notifyCheckpointSubsumedAsync(checkpointId);
+ } catch (RejectedExecutionException ex) {
+ // This may happen if the mailbox is closed. It means that the task is shutting
+ // down, so we just ignore it.
+ LOG.debug(
+ "Notify checkpoint subsume {} for {} ({}) was rejected by the mailbox",
+ checkpointId,
+ taskNameWithSubtask,
+ executionId);
+ } catch (Exception e) {
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ LOG.warn("Error while subsuming checkpoint {}.", checkpointId, e);
+ }
+ }
+ } else {
+ LOG.info(
+ "Ignoring checkpoint subsume notification for non-running task {}.",
+ taskNameWithSubtask);
+ }
+ }
Review comment:
You have resolved my previous comment about code duplication, but it still looks very similar as `notifyCheckpointAborted()` and `notifyCheckpointCompleted()`?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java
##########
@@ -100,6 +102,16 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
}
}
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ if (!closed) {
+ if (wrapped instanceof AbstractStreamOperator) {
+ ((AbstractStreamOperator<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ } else if (wrapped instanceof AbstractStreamOperatorV2) {
+ ((AbstractStreamOperatorV2<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
You are still adding this method to the public API (`AbstractStreamOperator/V2`). What were the drawbacks of @rkhachatryan 's suggestion of calling `stateBackend.notifyCheckpointSubsumed()` somewhere directly from the `StreamTask`/`OperatorChain`/`StreamOperatorWrapper`, bypassing operators all-together?
It's less consistent, but we could use `getKeyedStateBackend()` here, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885587084
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321 (Sat Aug 28 13:08:30 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678241150
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -229,6 +229,8 @@
private final ExecutionAttemptMappingProvider attemptMappingProvider;
+ @Nullable private CompletedCheckpoint lastSubsumedCheckpoint;
+
Review comment:
Thanks for your suggestion, I will consider to change this.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1220,8 +1224,15 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
Preconditions.checkState(pendingCheckpoint.isDisposed() && completedCheckpoint != null);
try {
- completedCheckpointStore.addCheckpoint(
- completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
+ CompletedCheckpoint lastSubsumed =
+ completedCheckpointStore.addCheckpointAndSubsumeOldestOne(
+ completedCheckpoint,
+ checkpointsCleaner,
+ this::scheduleTriggerRequest);
+ if (lastSubsumed != null && lastSubsumed.discardOnSubsume()) {
Review comment:
If we also send notification of subsumed savepoint, there would be a problem that we might subsume a savepoint and a checkpoint in the same turn when adding a new checkpoint. For this case, which checkpoint id should we notify?
I see your PR of https://github.com/apache/flink/pull/16575 also consider the savepoint case on TM side, was that the reason you prefer to send notification of savepoints?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677283450
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
I think we can refer to the documentation of `CompletedCheckpointStore#addCheckpoint`:
~~~ java
Only a bounded number of checkpoints is kept. When exceeding the maximum number of
retained checkpoints, the oldest one will be discarded.
~~~
And all current implementations obey this role to make this method atomic and discard the oldest checkpoint internally. That's why, to me, changing this method to return the oldest discarded checkpoint looks so straightforward.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678144570
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
Maybe we could record the latest subsumed checkpoint in the store, however, this might not get the core idea of current `CompletedCheckpointStore#addCheckpoint`.
You can take a look at `DefaultCompletedCheckpointStore#addCheckpoint`, once the `checkpointStateHandleStore.addAndLock(path, checkpoint)` failed, it would not trigger the subsume process anymore to ensure strict high-availability. The timing to subsume checkpoint is very important and just record a subsumed checkpoint but not known when that one is subsumed would lose information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678121291
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStore.java
##########
@@ -45,8 +47,12 @@
*
* <p>Only a bounded number of checkpoints is kept. When exceeding the maximum number of
* retained checkpoints, the oldest one will be discarded.
+ *
+ * @return the subsumed oldest completed checkpoint if possible, return null if no checkpoint
+ * needs to be discarded on subsume.
*/
- void addCheckpoint(
+ @Nullable
+ CompletedCheckpoint addCheckpointAndSubsumeOldestOne(
Review comment:
commented in the original [thread](https://github.com/apache/flink/pull/16582#discussion_r678121731).
~The new name doesn't seem to reflect what the method actually does.~
~It doesn't necessarily subsume **one** checkpoint (or **any** checkpoints at all).~
~The correct name would be `addAndMaybeSubsumeOldestAndReturnTheLatestSubsumedIfAny` - which clearly indicates to me that the new functionality should be in a separate method.~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
}, {
"hash" : "ab089745cc9841edb718c33a8ea3224775e38766",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ab089745cc9841edb718c33a8ea3224775e38766",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4575a58b2365b9ab070606108ef6ba1b4507dc24 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764)
* ab089745cc9841edb718c33a8ea3224775e38766 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677300054
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1399,6 +1399,34 @@ public void notifyCheckpointAborted(final long checkpointID) {
}
}
+ @Override
+ public void notifyCheckpointSubsumed(long checkpointId) {
+ AbstractInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
+
+ try {
+ invokable.notifyCheckpointSubsumedAsync(checkpointId);
+ } catch (RejectedExecutionException ex) {
+ // This may happen if the mailbox is closed. It means that the task is shutting
+ // down, so we just ignore it.
+ LOG.debug(
+ "Notify checkpoint subsume {} for {} ({}) was rejected by the mailbox",
+ checkpointId,
+ taskNameWithSubtask,
+ executionId);
+ } catch (Exception e) {
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ LOG.warn("Error while subsuming checkpoint {}.", checkpointId, e);
+ }
+ }
+ } else {
+ LOG.info(
+ "Ignoring checkpoint subsume notification for non-running task {}.",
+ taskNameWithSubtask);
+ }
+ }
+
Review comment:
Thanks for your suggestion, I will do the refactoring.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677287759
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener, KeyContext, Serializable {
Review comment:
Do we want to expose `notifyCheckpointSubsumed()` on the DataStream API level in a `@Public` class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r743366210
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
##########
@@ -339,78 +343,107 @@ public void checkpointState(
public void notifyCheckpointComplete(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
- Exception previousException = null;
- if (!isRunning.get()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for not-running task {}",
- checkpointId,
- taskName);
- } else if (operatorChain.isFinishedOnRestore()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for finished on restore task {}",
- checkpointId,
- taskName);
- } else {
- LOG.debug(
- "Notification of completed checkpoint {} for task {}", checkpointId, taskName);
- for (StreamOperatorWrapper<?, ?> operatorWrapper :
- operatorChain.getAllOperators(true)) {
- try {
- operatorWrapper.notifyCheckpointComplete(checkpointId);
- } catch (Exception e) {
- previousException = ExceptionUtils.firstOrSuppressed(e, previousException);
- }
- }
- }
- env.getTaskStateManager().notifyCheckpointComplete(checkpointId);
- ExceptionUtils.tryRethrowException(previousException);
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of complete checkpoint {} for not-running task {}",
+ "Ignoring notification of complete checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointComplete(checkpointId),
+ (tsm) -> tsm.notifyCheckpointComplete(checkpointId));
}
@Override
public void notifyCheckpointAborted(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of aborted checkpoint {} for not-running task {}",
+ "Ignoring notification of aborted checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ () -> {
+ boolean canceled = cancelAsyncCheckpointRunnable(checkpointId);
+
+ if (!canceled) {
+ if (checkpointId > lastCheckpointId) {
+ // only record checkpoints that have not triggered on task side.
+ abortedCheckpointIds.add(checkpointId);
+ }
+ }
+
+ channelStateWriter.abort(
+ checkpointId,
+ new CancellationException("checkpoint aborted via notification"),
+ false);
+ },
+ (opw) -> opw.getStreamOperator().notifyCheckpointAborted(checkpointId),
+ (tsm) -> tsm.notifyCheckpointAborted(checkpointId));
+ }
+
+ @Override
+ public void notifyCheckpointSubsumed(
+ long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
+ throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of subsumed checkpoint {} for not-running task {}",
+ "Ignoring notification of subsumed checkpoint {} for finished on restore task {}",
+ "Notification of subsumed checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointSubsumed(checkpointId),
+ null);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ OperatorChain<?, ?> operatorChain,
+ Supplier<Boolean> isRunning,
+ String logFormatIfNotRunning,
+ String logFormatIfFinishedOnRestore,
+ String logFormatIfNotificationStart,
+ @Nullable Runnable runnableBeforeNotifyOperator,
+ @Nonnull ThrowingConsumer<StreamOperatorWrapper<?, ?>, Exception> operatorConsumer,
+ @Nullable ThrowingConsumer<TaskStateManager, Exception> taskStateManagerConsumer)
+ throws Exception {
Review comment:
The refactor also take the `enum` of `NotifyCheckpointOperation` as `Task` did.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677318208
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
Yes, it's clear when one goes through the documentation.
However, I think it's better to make the API clear even without the need to read the javadoc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5817be659ecb4b6124f01415e552de29f538960b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030)
* 36cfdc4abcad6028059c134f323e963d455921e8 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678133482
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1220,8 +1224,15 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
Preconditions.checkState(pendingCheckpoint.isDisposed() && completedCheckpoint != null);
try {
- completedCheckpointStore.addCheckpoint(
- completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
+ CompletedCheckpoint lastSubsumed =
+ completedCheckpointStore.addCheckpointAndSubsumeOldestOne(
+ completedCheckpoint,
+ checkpointsCleaner,
+ this::scheduleTriggerRequest);
+ if (lastSubsumed != null && lastSubsumed.discardOnSubsume()) {
Review comment:
I think this needs more discussion here. After we store savepoint in checkpoint store, the actual meaning of `CompletedCheckpointStore#addCheckpoint` has changed: we might subsume a savepoint and a checkpoint in the same turn when adding a new checkpoint. My understanding is that we would not need to notify tasks of subsumed savepoint as it should not be owned on task side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678215722
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1220,8 +1224,15 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
Preconditions.checkState(pendingCheckpoint.isDisposed() && completedCheckpoint != null);
try {
- completedCheckpointStore.addCheckpoint(
- completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
+ CompletedCheckpoint lastSubsumed =
+ completedCheckpointStore.addCheckpointAndSubsumeOldestOne(
+ completedCheckpoint,
+ checkpointsCleaner,
+ this::scheduleTriggerRequest);
+ if (lastSubsumed != null && lastSubsumed.discardOnSubsume()) {
Review comment:
I think we shouldn't make such a distinction.
If TM needs to distinguish for some particular task (state management), it can be done on TM side.
One practical reason to notify TMs is that subsuming a savepoint allows them subsume older checkpoints.
A more general reason that the community I think is moving towards miminizing the difference in handling of savepoints and checkpoints.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -229,6 +229,8 @@
private final ExecutionAttemptMappingProvider attemptMappingProvider;
+ @Nullable private CompletedCheckpoint lastSubsumedCheckpoint;
+
Review comment:
I think the added complexity in `CheckpointCoordinator` doesn't worth it as it's a critical component.
And it's more valuable to test the actual functionality - i.e. what ID is sent in the notifications.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677217866
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
##########
@@ -136,6 +136,11 @@ public void notifyCheckpointAborted(long checkpointId) throws Exception {
}
}
+ @Override
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ super.notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
Hmm, my fault. Since current `userFunction` is not instance of `InternalCheckpointListener`, we could remove this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677299128
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener, KeyContext, Serializable {
Review comment:
@pnowojski I think this needs more discussion here.
My previous idea was to add a default no-op in `CheckpointListener` so that could be public. Roman convinced me that this new interface does not make so much senses for most users, and could be acted as an internal API just like what `InternalKvState` did.
However, current PR has to change several places to make the logic more clean, which makes me to think whether previous `AbstractStreamOperator` did not welcome such internal semantics here.
The main usage of `notifyCheckpointSubsumed` is for abstract keyed state backend, we can compare the relationship of `AbstractStreamOperator` with `AbstractKeyedStateBackend`. If `AbstractStreamOperator` is open for users to implement and `AbstractKeyedStateBackend` is actually also open for users to implement their own state backend.
If so, we might treat `notifyCheckpointSubsumed` as public also.
The only left questions are can we add a new method in `@Public` class and we need to remember to implement it in necessary places.
cc @rkhachatryan
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677193306
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
Actually, current `CheckpointStore#addCheckpoint` has the meaning that this is an atomic operation which will add the completed checkpoint and subsumed an older checkpoint. I think return the subsumed checkpoint is clearly for this operation.
If we adopt the API `long getLatestSubsumedCheckpointID()`, we will miss some information that when the checkpoint is subsumed and we might also need to check whether the latest subsumed checkpoint ID has been changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r676720289
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
Why don't we just remember the latest subsumed checkpoint in Store (which actually subsumes them) and let the Coordinator ask for it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10466b2d0065049cc8d003cdfc1395785d279e23 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979)
* 5817be659ecb4b6124f01415e552de29f538960b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678245161
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
> Yes, I agree that the returned subsumed checkpoint might be null
From the (current) name, it's unclear which checkpoint is returned, even if not null.
> I can't see why we need to add a separate method?
> who will call the original addCheckpoint method?
In my opinion, it would make a more clear API.
`CheckpointCoordinator` will call it (same as now). Let me demonstrate this:
https://github.com/rkhachatryan/flink/commit/10242b7aef2f4f2db94c924f940928fce4ab0fc9#diff-2c7004a2d412c3566de5ff6fb9e6d027742328c2acad60685e47ce4ba9df0810R1286
WDYT?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
WDYT about [this](https://github.com/rkhachatryan/flink/commit/10242b7aef2f4f2db94c924f940928fce4ab0fc9#diff-1bc8de30a2deacb9128e6bb8c13169b3cb692beddb08e6724f1cf2b550435effR264) approach?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1220,8 +1224,15 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
Preconditions.checkState(pendingCheckpoint.isDisposed() && completedCheckpoint != null);
try {
- completedCheckpointStore.addCheckpoint(
- completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
+ CompletedCheckpoint lastSubsumed =
+ completedCheckpointStore.addCheckpointAndSubsumeOldestOne(
+ completedCheckpoint,
+ checkpointsCleaner,
+ this::scheduleTriggerRequest);
+ if (lastSubsumed != null && lastSubsumed.discardOnSubsume()) {
Review comment:
> which checkpoint id should we notify?
I think it should always be the latest (regardless whether it's savepoint or checkpoint).
> I see your PR of #16575 also consider the savepoint case on TM side, was that the reason you prefer to send notification of savepoints?
No, I mentioned the reasons above (the main one is generality).
Regarding the PR (#16575), TM needs to distinguish when discarding the state anyways. This is because it may have non-confirmed savepoints (that might be aborted); at the same, a newer checkpoint might be subsumed; which would mean that those older savepoints are subsumed.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1220,8 +1224,15 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
Preconditions.checkState(pendingCheckpoint.isDisposed() && completedCheckpoint != null);
try {
- completedCheckpointStore.addCheckpoint(
- completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
+ CompletedCheckpoint lastSubsumed =
+ completedCheckpointStore.addCheckpointAndSubsumeOldestOne(
+ completedCheckpoint,
+ checkpointsCleaner,
+ this::scheduleTriggerRequest);
+ if (lastSubsumed != null && lastSubsumed.discardOnSubsume()) {
Review comment:
> which checkpoint id should we notify?
I think it should always be the latest (regardless of whether it's savepoint or checkpoint).
> I see your PR of #16575 also consider the savepoint case on TM side, was that the reason you prefer to send notification of savepoints?
No, I mentioned the reasons above (the main one is generality).
Regarding the PR (#16575), TM needs to distinguish when discarding the state anyways. This is because it may have non-confirmed savepoints (that might be aborted); at the same, a newer checkpoint might be subsumed; which would mean that those older savepoints are subsumed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 935abc391163053cdc2d5fb35d5f321844f0e8b1 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 935abc391163053cdc2d5fb35d5f321844f0e8b1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886)
* 10466b2d0065049cc8d003cdfc1395785d279e23 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10466b2d0065049cc8d003cdfc1395785d279e23 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979)
* 5817be659ecb4b6124f01415e552de29f538960b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10466b2d0065049cc8d003cdfc1395785d279e23 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 36cfdc4abcad6028059c134f323e963d455921e8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078)
* dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r676745053
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener, KeyContext, Serializable {
Review comment:
It seems wrong to me that a `@PublicEvolving` interface extends `@Internal` interface.
The root cause I think is in the `StreamOperator` hierarchy (but that's a different story).
Semantically, I think operators should only be aware of checkpoint completion (to emit data for example); their backends should in addition know about the subsumption.
I see the following solutions to notify state backends about the subsumption
1. Modify `CheckpointListener` interface instead of introducing `InternalCheckpointListener` as you proposed initially
2. In `StreamOperatorWrapper`, get `KeyedStateBackend` from the operator (if it's `AbstractStreamOperator`) and notify it directly (if it's `InternalCheckpointListener`)
3. Same, but via `StreamOperatorStateHandler`
WDYT?
cc: @pnowojski
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4575a58b2365b9ab070606108ef6ba1b4507dc24 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r754394153
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1405,59 +1406,80 @@ private void declineCheckpoint(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final TaskInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING) {
- checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
- try {
- ((CheckpointableTask) invokable).notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.COMPLETE);
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final TaskInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID, latestCompletedCheckpointId, NotifyCheckpointOperation.ABORT);
+ }
- if (executionState == ExecutionState.RUNNING) {
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.SUBSUME);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ long latestCompletedCheckpointId,
+ NotifyCheckpointOperation notifyCheckpointOperation) {
+ TaskInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
try {
- ((CheckpointableTask) invokable)
- .notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointAbortAsync(
+ checkpointId, latestCompletedCheckpointId);
+ break;
+ case COMPLETE:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointCompleteAsync(checkpointId);
+ break;
+ case SUBSUME:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointSubsumedAsync(checkpointId);
+ }
} catch (RejectedExecutionException ex) {
// This may happen if the mailbox is closed. It means that the task is shutting
// down, so we just ignore it.
LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
+ "Notify checkpoint {}} {} for {} ({}) was rejected by the mailbox.",
+ notifyCheckpointOperation,
+ checkpointId,
taskNameWithSubtask,
executionId);
} catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ case COMPLETE:
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ failExternally(
+ new RuntimeException(
+ String.format(
+ "Error while notify checkpoint %s.",
+ notifyCheckpointOperation),
+ t));
+ }
+ break;
+ case SUBSUME:
+ // just rethrow the throwable out as we do not expect notification of
+ // subsume could fail the task.
+ ExceptionUtils.rethrow(t);
Review comment:
I'm a little bit thorn apart here.
Wouldn't the same logic apply to `notifyCheckpointAbort()`?
On the other hand, even in the case of an exception during `notifyCheckpointComplete()`. As long as checkpoint is failed, there won't be any data loss. Data will be committed in the next checkpoint OR if there are enough checkpoint failures, the whole job will failover because of the exceeded number of tolerable checkpoint failures.
I think it could be argued both directions for either of those methods (abort/complete/subsume), so maybe for the sake of simplicity and consistency it would be best to keep them behaving in the same way?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r745617126
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -46,18 +46,24 @@
class CheckpointSubsumeHelper {
private static final Logger LOG = LoggerFactory.getLogger(CheckpointSubsumeHelper.class);
- public static void subsume(
+ public static Optional<CompletedCheckpoint> subsume(
Deque<CompletedCheckpoint> checkpoints, int numRetain, SubsumeAction subsumeAction)
throws Exception {
if (checkpoints.isEmpty() || checkpoints.size() <= numRetain) {
- return;
+ return Optional.empty();
}
CompletedCheckpoint latest = checkpoints.peekLast();
+ CompletedCheckpoint lastSubsumedCheckpoint = null;
Review comment:
optional nit: `Optional<CompletedCheckpoint> lastSubsumedCheckpoint = Optional.empty();`?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1405,59 +1406,80 @@ private void declineCheckpoint(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final TaskInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING) {
- checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
- try {
- ((CheckpointableTask) invokable).notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.COMPLETE);
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final TaskInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID, latestCompletedCheckpointId, NotifyCheckpointOperation.ABORT);
+ }
- if (executionState == ExecutionState.RUNNING) {
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.SUBSUME);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ long latestCompletedCheckpointId,
+ NotifyCheckpointOperation notifyCheckpointOperation) {
+ TaskInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
try {
- ((CheckpointableTask) invokable)
- .notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointAbortAsync(
+ checkpointId, latestCompletedCheckpointId);
+ break;
+ case COMPLETE:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointCompleteAsync(checkpointId);
+ break;
+ case SUBSUME:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointSubsumedAsync(checkpointId);
+ }
} catch (RejectedExecutionException ex) {
// This may happen if the mailbox is closed. It means that the task is shutting
// down, so we just ignore it.
LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
+ "Notify checkpoint {}} {} for {} ({}) was rejected by the mailbox.",
+ notifyCheckpointOperation,
+ checkpointId,
taskNameWithSubtask,
executionId);
} catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ case COMPLETE:
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ failExternally(
+ new RuntimeException(
+ String.format(
+ "Error while notify checkpoint %s.",
+ notifyCheckpointOperation),
+ t));
+ }
+ break;
+ case SUBSUME:
+ // just rethrow the throwable out as we do not expect notification of
+ // subsume could fail the task.
+ ExceptionUtils.rethrow(t);
Review comment:
Why do we need a special handling here?
What do you mean "We do not expect notification could fail the task"? Do you mean that this error should never happen? If so, I would keep the error handling unified regardless.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 36cfdc4abcad6028059c134f323e963d455921e8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678136476
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
Yes, I agree that the returned subsumed checkpoint might be null, but I can't see why we need to add a separate method? If so, who will call the original `addCheckpoint` method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-888053395
@rkhachatryan @pnowojski , thanks for your review. I have refactored this PR with two commits:
1beb902: Refactor `CompletedCheckpointStore#addCheckpoint` to `CompletedCheckpoint#addCheckpointAndSubsumeOldestOne`.
36cfdc4: Introduce the notification of subsumed checkpoint which addressed your comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r676865635
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener, KeyContext, Serializable {
Review comment:
I meant to add a method `checkpointSubsumed()` to the `CheckpointListener` (or `lastSubsumedId` argument to its existing `notifyCheckpointComplete()` method).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678130044
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -229,6 +229,8 @@
private final ExecutionAttemptMappingProvider attemptMappingProvider;
+ @Nullable private CompletedCheckpoint lastSubsumedCheckpoint;
+
Review comment:
This is just for easy of testing in `CheckpointCoordinatorTest`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677176343
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
I think returning **subsumed** checkpoint from `addCheckpoint` is not very straightforward (as a caller, I'd expect **added** checkpoint to be returned).
I'd rather add `long getLatestSubsumedCheckpointID();` method or something like that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r676716548
##########
File path: flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.java
##########
@@ -0,0 +1,52 @@
+/*
+
+* 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.
+
+*/
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * The internal checkpoint listener add another {@link #notifyCheckpointSubsumed(long)} interface
+ * for internal usage.
+ */
+@Internal
+public interface InternalCheckpointListener extends CheckpointListener {
+
+ /**
+ * A invalid checkpoint id, which could be used in {@link #notifyCheckpointSubsumed(long)} to
+ * represent no checkpoint needs to be subsumed.
+ */
+ long INVALID_CHECKPOINT_ID = -1;
+
+ /**
+ * This method is called as a notification once a distributed checkpoint has been subsumed.
+ *
+ * <p>These notifications are "best effort", meaning they can sometimes be skipped.
+ *
+ * <p>This method is very rarely necessary to implement. The "best effort" guarantee, together
+ * with the fact that this method should not result in discarding any data (per the "Checkpoint
+ * Subsuming Contract") means it is mainly useful for earlier cleanups of auxiliary resources.
+ *
+ * @param checkpointId The ID of the checkpoint that has been subsumed.
+ * @throws Exception This method can propagate exceptions, which leads to a failure/recovery for
+ * the task or job.
+ */
+ default void notifyCheckpointSubsumed(long checkpointId) throws Exception {}
Review comment:
I think having a default no-op implementation is error-prone (as it allows to forget to implement it). Do we really need it?
edit:
as I commented below `AbstractStreamOperatorV2` misses the implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r681106908
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java
##########
@@ -100,6 +102,16 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
}
}
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ if (!closed) {
+ if (wrapped instanceof AbstractStreamOperator) {
+ ((AbstractStreamOperator<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ } else if (wrapped instanceof AbstractStreamOperatorV2) {
+ ((AbstractStreamOperatorV2<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
That's a tough call. It should follow the hierarchy and `StreamOperator` shouldn't be the public API (we are planning to remove it from the public api at some point). But as it is, it doesn't matter if you add this method to `StreamOperator` or `AbstractStreamOperator`, because all practical cases for implementing a custom `StreamOperator` are always extending from `AbstractStreamOperator`. So I would propose to call `notifyCheckpointSubsumed()` via `getKeyedStateBackend()` for now. Later, once we remove `StreamOperator` from public API we can change it back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r755992249
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1405,59 +1406,80 @@ private void declineCheckpoint(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final TaskInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING) {
- checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
- try {
- ((CheckpointableTask) invokable).notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.COMPLETE);
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final TaskInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID, latestCompletedCheckpointId, NotifyCheckpointOperation.ABORT);
+ }
- if (executionState == ExecutionState.RUNNING) {
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.SUBSUME);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ long latestCompletedCheckpointId,
+ NotifyCheckpointOperation notifyCheckpointOperation) {
+ TaskInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
try {
- ((CheckpointableTask) invokable)
- .notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointAbortAsync(
+ checkpointId, latestCompletedCheckpointId);
+ break;
+ case COMPLETE:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointCompleteAsync(checkpointId);
+ break;
+ case SUBSUME:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointSubsumedAsync(checkpointId);
+ }
} catch (RejectedExecutionException ex) {
// This may happen if the mailbox is closed. It means that the task is shutting
// down, so we just ignore it.
LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
+ "Notify checkpoint {}} {} for {} ({}) was rejected by the mailbox.",
+ notifyCheckpointOperation,
+ checkpointId,
taskNameWithSubtask,
executionId);
} catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ case COMPLETE:
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ failExternally(
+ new RuntimeException(
+ String.format(
+ "Error while notify checkpoint %s.",
+ notifyCheckpointOperation),
+ t));
+ }
+ break;
+ case SUBSUME:
+ // just rethrow the throwable out as we do not expect notification of
+ // subsume could fail the task.
+ ExceptionUtils.rethrow(t);
Review comment:
From my point of view, our internal data structures are expected to keep working fine if come across an exception here. Some customized keyed state backend might introduce IO operations within `notifyCheckpointSubsumed()`, and the task should not fail as those IO operations are best-effort-to-do. That's why my implementation would not fail here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r742598361
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
@pnowojski , thanks for your review! Sorry for late reply due to recent too busy internal work.
The reason I prefer the solution to return the checkpoint to subsume is that we might not susbume checkpoint each time especially when we adding a complete savepoint (see [FLINK-21351](https://issues.apache.org/jira/browse/FLINK-21351)). Just use a getter would lost such information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474)
* 8ca91f13df7b3d17d2137e008e51acbcb4067c97 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r680710358
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1399,6 +1399,33 @@ public void notifyCheckpointAborted(final long checkpointID) {
}
}
+ public void notifyCheckpointSubsumed(long checkpointId) {
+ AbstractInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
+
+ try {
+ invokable.notifyCheckpointSubsumedAsync(checkpointId);
+ } catch (RejectedExecutionException ex) {
+ // This may happen if the mailbox is closed. It means that the task is shutting
+ // down, so we just ignore it.
+ LOG.debug(
+ "Notify checkpoint subsume {} for {} ({}) was rejected by the mailbox",
+ checkpointId,
+ taskNameWithSubtask,
+ executionId);
+ } catch (Exception e) {
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ LOG.warn("Error while subsuming checkpoint {}.", checkpointId, e);
+ }
+ }
+ } else {
+ LOG.info(
+ "Ignoring checkpoint subsume notification for non-running task {}.",
+ taskNameWithSubtask);
+ }
+ }
Review comment:
I mistakenly thought that you left the comments on `SubtaskCheckpointCoordinatorImpl` class, and I refactored that class to remove code duplication. And after your remider, I see your comments are under `Task` class, and I will refactor later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r680716956
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java
##########
@@ -100,6 +102,16 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
}
}
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ if (!closed) {
+ if (wrapped instanceof AbstractStreamOperator) {
+ ((AbstractStreamOperator<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ } else if (wrapped instanceof AbstractStreamOperatorV2) {
+ ((AbstractStreamOperatorV2<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
Adding the method was to just follow operator hierarchy and we could extend the usage to other componments ont only keyed state backends if necessary in the future.
If you don't like to add methods in `PublicEvolving` class, I think it's fine and I could revert to call `getKeyedStateBackend()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r680673722
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java
##########
@@ -100,6 +102,16 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
}
}
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ if (!closed) {
+ if (wrapped instanceof AbstractStreamOperator) {
+ ((AbstractStreamOperator<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ } else if (wrapped instanceof AbstractStreamOperatorV2) {
+ ((AbstractStreamOperatorV2<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
You are still adding this method to the public API (`AbstractStreamOperator/V2`). What were the drawbacks of @rkhachatryan 's suggestion of calling `stateBackend.notifyCheckpointSubsumed()` somewhere directly from the `StreamTask`/`OperatorChain`/`StreamOperatorWrapper`, bypassing operators all-together?
It's less consistent, but we could use `getKeyedStateBackend()` here as @rkhachatryan suggested, right? Or what is wrong with this approach?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r676851147
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener, KeyContext, Serializable {
Review comment:
> It seems wrong to me that a @PublicEvolving interface extends @Internal interface.
+1
What do you @rkhachatryan mean by:
> 1. Modify CheckpointListener interface instead of introducing InternalCheckpointListener as you proposed initially
?
Regardless I also think we shouldn't go through the operators if we don't want to make `notifyCheckpointSubsumed()` public.
##########
File path: flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.java
##########
@@ -0,0 +1,52 @@
+/*
+
+* 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.
+
+*/
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * The internal checkpoint listener add another {@link #notifyCheckpointSubsumed(long)} interface
+ * for internal usage.
+ */
+@Internal
+public interface InternalCheckpointListener extends CheckpointListener {
+
+ /**
+ * A invalid checkpoint id, which could be used in {@link #notifyCheckpointSubsumed(long)} to
+ * represent no checkpoint needs to be subsumed.
+ */
+ long INVALID_CHECKPOINT_ID = -1;
+
+ /**
+ * This method is called as a notification once a distributed checkpoint has been subsumed.
+ *
+ * <p>These notifications are "best effort", meaning they can sometimes be skipped.
+ *
+ * <p>This method is very rarely necessary to implement. The "best effort" guarantee, together
+ * with the fact that this method should not result in discarding any data (per the "Checkpoint
+ * Subsuming Contract") means it is mainly useful for earlier cleanups of auxiliary resources.
+ *
+ * @param checkpointId The ID of the checkpoint that has been subsumed.
+ * @throws Exception This method can propagate exceptions, which leads to a failure/recovery for
+ * the task or job.
+ */
+ default void notifyCheckpointSubsumed(long checkpointId) throws Exception {}
Review comment:
Yes I second this comment.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1399,6 +1399,34 @@ public void notifyCheckpointAborted(final long checkpointID) {
}
}
+ @Override
+ public void notifyCheckpointSubsumed(long checkpointId) {
+ AbstractInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
+
+ try {
+ invokable.notifyCheckpointSubsumedAsync(checkpointId);
+ } catch (RejectedExecutionException ex) {
+ // This may happen if the mailbox is closed. It means that the task is shutting
+ // down, so we just ignore it.
+ LOG.debug(
+ "Notify checkpoint subsume {} for {} ({}) was rejected by the mailbox",
+ checkpointId,
+ taskNameWithSubtask,
+ executionId);
+ } catch (Exception e) {
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ LOG.warn("Error while subsuming checkpoint {}.", checkpointId, e);
+ }
+ }
+ } else {
+ LOG.info(
+ "Ignoring checkpoint subsume notification for non-running task {}.",
+ taskNameWithSubtask);
+ }
+ }
+
Review comment:
Maybe try to de-duplicate the code? Existing `notifyCheckpointXXX` looks the same?
##########
File path: flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.java
##########
@@ -0,0 +1,52 @@
+/*
+
+* 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.
+
+*/
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * The internal checkpoint listener add another {@link #notifyCheckpointSubsumed(long)} interface
+ * for internal usage.
+ */
+@Internal
+public interface InternalCheckpointListener extends CheckpointListener {
Review comment:
Do we need this interface in the first place? The only value I can see of this is to have a common signature of this `notifyCheckpointSubsumed()`, which maybe is a good reason on it's own, which would make refactoring/renaming easier, but I'm not entirely sure...
Also I'm not entirely sure if `InternalCheckpointListener` should extend `CheckpointListener` or rather we should have another smaller interface `CheckpointSubsumedListener` that just adds this single method.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -138,7 +138,7 @@
TaskSlotPayload,
TaskActions,
PartitionProducerStateProvider,
- CheckpointListener {
+ InternalCheckpointListener {
Review comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410)
* 8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677345113
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener, KeyContext, Serializable {
Review comment:
I'm also not sure if it makes sense to expose `notifyCheckpointSubsumed()` to the users. And if we don't have a good motivating example, I would prefer to not expose it. Why clog the public API at best? And at worst this could cause us problems down the road for providing backward compatibility/refactoring. If needed, we can always add this notification later. Removing and maintaining things is harder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r676745053
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener, KeyContext, Serializable {
Review comment:
It seems wrong to me that a `@PublicEvolving` interface extends `@Internal` interface.
The root cause I think is in the `StreamOperator` hierarchy (but that's a different story).
Semantically, I think operators should only be aware of checkpoint completion (to emit data for example); their backends should in addition know about the subsumption.
I see the following solutions to notify state backends about the subsumption
1. Modify `CheckpointListener` interface instead of introducing `InternalCheckpointListener` as you proposed initially
2. In `StreamOperatorWrapper`, get `KeyedStateBackend` from the operator (if it's `AbstractStreamOperator`) and notify it directly (if it's `InternalCheckpointListener`):
```
// SubtaskCheckpointCoordinatorImpl.notifyCheckpointSubsumed:
// loop through the chain:
operatorWrapper.notifyCheckpointSubsumed(checkpointId);
// StreamOperatorWrapper.notifyCheckpointSubsumed:
// .. casts omitted ...
`getStreamOperator().getKeyedStateBackend().notifyCheckpointSubsumed(checkpointId);`
```
3. Same, but via `StreamOperatorStateHandler`
WDYT?
cc: @pnowojski
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677174708
##########
File path: flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.java
##########
@@ -0,0 +1,52 @@
+/*
+
+* 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.
+
+*/
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * The internal checkpoint listener add another {@link #notifyCheckpointSubsumed(long)} interface
+ * for internal usage.
+ */
+@Internal
+public interface InternalCheckpointListener extends CheckpointListener {
+
+ /**
+ * A invalid checkpoint id, which could be used in {@link #notifyCheckpointSubsumed(long)} to
+ * represent no checkpoint needs to be subsumed.
+ */
+ long INVALID_CHECKPOINT_ID = -1;
+
+ /**
+ * This method is called as a notification once a distributed checkpoint has been subsumed.
+ *
+ * <p>These notifications are "best effort", meaning they can sometimes be skipped.
+ *
+ * <p>This method is very rarely necessary to implement. The "best effort" guarantee, together
+ * with the fact that this method should not result in discarding any data (per the "Checkpoint
+ * Subsuming Contract") means it is mainly useful for earlier cleanups of auxiliary resources.
+ *
+ * @param checkpointId The ID of the checkpoint that has been subsumed.
+ * @throws Exception This method can propagate exceptions, which leads to a failure/recovery for
+ * the task or job.
+ */
+ default void notifyCheckpointSubsumed(long checkpointId) throws Exception {}
Review comment:
The original idea is to like current `CheckpointListener#notifyCheckpointAborted` which makes a default implementation to not let the normal users know. After making the interface not a default no-op implementation, I tried to make those abstract class implement this necessary interface to make the changes as least as possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677324317
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
I see, maybe the previous name of `addCheckpoint` is not clear for what it really does. It actually does `addCheckpointAndSubsumeOldestOne`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678144570
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
Maybe we could record the latest subsumed checkpoint in the store, however, this might not get the core idea of current `CompletedCheckpointStore#addCheckpoint`.
You can take a look at `DefaultCompletedCheckpointStore#addCheckpoint`, once the `checkpointStateHandleStore.addAndLock(path, checkpoint)` failed, it would trigger the subsume process anymore to ensure strict high-availability. The timing to subsume checkpoint is very important and just record a subsumed checkpoint but not known when that one is subsumed would lose information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678144570
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
Maybe we could record the latest subsumed checkpoint in the store, however, this might not get the core idea of current `CompletedCheckpointStore#addCheckpoint`.
You can take a look at `DefaultCompletedCheckpointStore#addCheckpoint`, once the `checkpointStateHandleStore.addAndLock(path, checkpoint)` failed, it would not trigger the subsume process anymore to ensure strict high-availability. The timing to subsume checkpoint is very important and just record a subsumed checkpoint but not know when that one is subsumed would lose information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r677170632
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
The original idea is to not change the API, and I think a proper place is to change the return result of `CompletedCheckpointStore#addCheckpoint` from `void` to the subsumed `CompletedCheckpoint`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r676720289
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
Why don't we just remember the latest subsumed checkpoint in store (which actually subsumes them) and let coordinator ask for it?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -138,7 +138,7 @@
TaskSlotPayload,
TaskActions,
PartitionProducerStateProvider,
- CheckpointListener {
+ InternalCheckpointListener {
Review comment:
I think it's not neccessary to use the interface (`InternalCheckpointListener`) here as the caller already knows the type exactly (`Task`).
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
##########
@@ -136,6 +136,11 @@ public void notifyCheckpointAborted(long checkpointId) throws Exception {
}
}
+ @Override
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ super.notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
Is this overriding actually necessary?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext, Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener, KeyContext, Serializable {
Review comment:
It seems wrong that a `@PublicEvolving` interface extends `@Internal` interface.
The root cause I think in the `StreamOperator` hierarchy, but that's a long story.
Semantically, I think operators should only be aware of checkpoint completion (to emit data for example); their backends should in addition know about the subsumption.
I see the following solutions to notify state backends about the subsumption
1. Modify `CheckpointListener` interface instead of introducing `InternalCheckpointListener` as you proposed initially
2. In `StreamOperatorWrapper`, get `KeyedStateBackend` from the operator (if it's `AbstractStreamOperator`) and notify it directly (if it's `InternalCheckpointListener`)
3. Same, but via `StreamOperatorStateHandler`
WDYT?
cc: @pnowojski
##########
File path: flink-core/src/main/java/org/apache/flink/api/common/state/InternalCheckpointListener.java
##########
@@ -0,0 +1,52 @@
+/*
+
+* 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.
+
+*/
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * The internal checkpoint listener add another {@link #notifyCheckpointSubsumed(long)} interface
+ * for internal usage.
+ */
+@Internal
+public interface InternalCheckpointListener extends CheckpointListener {
+
+ /**
+ * A invalid checkpoint id, which could be used in {@link #notifyCheckpointSubsumed(long)} to
+ * represent no checkpoint needs to be subsumed.
+ */
+ long INVALID_CHECKPOINT_ID = -1;
+
+ /**
+ * This method is called as a notification once a distributed checkpoint has been subsumed.
+ *
+ * <p>These notifications are "best effort", meaning they can sometimes be skipped.
+ *
+ * <p>This method is very rarely necessary to implement. The "best effort" guarantee, together
+ * with the fact that this method should not result in discarding any data (per the "Checkpoint
+ * Subsuming Contract") means it is mainly useful for earlier cleanups of auxiliary resources.
+ *
+ * @param checkpointId The ID of the checkpoint that has been subsumed.
+ * @throws Exception This method can propagate exceptions, which leads to a failure/recovery for
+ * the task or job.
+ */
+ default void notifyCheckpointSubsumed(long checkpointId) throws Exception {}
Review comment:
I think having a default no-op implementation is error-prone (as it allows to forget to implement it). Do we really need it?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManager.java
##########
@@ -42,7 +42,7 @@
* <p>This interface also offers the complementary method that provides access to previously saved
* state of operator instances in the task for restore purposes.
*/
-public interface TaskStateManager extends CheckpointListener, AutoCloseable {
+public interface TaskStateManager extends InternalCheckpointListener, AutoCloseable {
Review comment:
What is the intention here, is it `LocalStateStore` notification?
But as of now, `LocalStateStore` doesn't accept these notifications.
Other than that, I think `TaskStateManager` doesn't need to know about the subsumption.
I also couldn't find any call sites.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
##########
@@ -382,6 +382,11 @@ public void notifyCheckpointAborted(long checkpointId) throws Exception {
stateHandler.notifyCheckpointAborted(checkpointId);
}
+ @Override
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ stateHandler.notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
There is also `AbstractStreamOperatorV2`...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r755288365
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1405,59 +1406,80 @@ private void declineCheckpoint(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final TaskInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING) {
- checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
- try {
- ((CheckpointableTask) invokable).notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.COMPLETE);
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final TaskInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID, latestCompletedCheckpointId, NotifyCheckpointOperation.ABORT);
+ }
- if (executionState == ExecutionState.RUNNING) {
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.SUBSUME);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ long latestCompletedCheckpointId,
+ NotifyCheckpointOperation notifyCheckpointOperation) {
+ TaskInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
try {
- ((CheckpointableTask) invokable)
- .notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointAbortAsync(
+ checkpointId, latestCompletedCheckpointId);
+ break;
+ case COMPLETE:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointCompleteAsync(checkpointId);
+ break;
+ case SUBSUME:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointSubsumedAsync(checkpointId);
+ }
} catch (RejectedExecutionException ex) {
// This may happen if the mailbox is closed. It means that the task is shutting
// down, so we just ignore it.
LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
+ "Notify checkpoint {}} {} for {} ({}) was rejected by the mailbox.",
+ notifyCheckpointOperation,
+ checkpointId,
taskNameWithSubtask,
executionId);
} catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ case COMPLETE:
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ failExternally(
+ new RuntimeException(
+ String.format(
+ "Error while notify checkpoint %s.",
+ notifyCheckpointOperation),
+ t));
+ }
+ break;
+ case SUBSUME:
+ // just rethrow the throwable out as we do not expect notification of
+ // subsume could fail the task.
+ ExceptionUtils.rethrow(t);
Review comment:
I see, for the simplicity I will refactor this to keep the same behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r755767863
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1405,59 +1406,80 @@ private void declineCheckpoint(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final TaskInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING) {
- checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
- try {
- ((CheckpointableTask) invokable).notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.COMPLETE);
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final TaskInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID, latestCompletedCheckpointId, NotifyCheckpointOperation.ABORT);
+ }
- if (executionState == ExecutionState.RUNNING) {
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ CheckpointStoreUtil.INVALID_CHECKPOINT_ID,
+ NotifyCheckpointOperation.SUBSUME);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ long latestCompletedCheckpointId,
+ NotifyCheckpointOperation notifyCheckpointOperation) {
+ TaskInvokable invokable = this.invokable;
+
+ if (executionState == ExecutionState.RUNNING && invokable != null) {
checkState(invokable instanceof CheckpointableTask, "invokable is not checkpointable");
try {
- ((CheckpointableTask) invokable)
- .notifyCheckpointAbortAsync(checkpointID, latestCompletedCheckpointId);
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointAbortAsync(
+ checkpointId, latestCompletedCheckpointId);
+ break;
+ case COMPLETE:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointCompleteAsync(checkpointId);
+ break;
+ case SUBSUME:
+ ((CheckpointableTask) invokable)
+ .notifyCheckpointSubsumedAsync(checkpointId);
+ }
} catch (RejectedExecutionException ex) {
// This may happen if the mailbox is closed. It means that the task is shutting
// down, so we just ignore it.
LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected by the mailbox",
- checkpointID,
+ "Notify checkpoint {}} {} for {} ({}) was rejected by the mailbox.",
+ notifyCheckpointOperation,
+ checkpointId,
taskNameWithSubtask,
executionId);
} catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting checkpoint", t));
+ switch (notifyCheckpointOperation) {
+ case ABORT:
+ case COMPLETE:
+ if (getExecutionState() == ExecutionState.RUNNING) {
+ failExternally(
+ new RuntimeException(
+ String.format(
+ "Error while notify checkpoint %s.",
+ notifyCheckpointOperation),
+ t));
+ }
+ break;
+ case SUBSUME:
+ // just rethrow the throwable out as we do not expect notification of
+ // subsume could fail the task.
+ ExceptionUtils.rethrow(t);
Review comment:
Now I remembered some past discussion about the behaviour on failing tasks or not during checkpointing. Our previous conclusion was indeed no to fail the job if not needed. We have to fail the job, if the exception happened in for example the sync phase of snapshotting, as otherwise operators/statebackends/user functions might end up in an inconsistent state. However failures from for example async phase can be handled by the `CheckpointCoordinator` and `CheckpointFailureManager`.
`notifyCheckpointComplete` and `notifyCheckpointAborted` has to fail the job, at least in some/most cases, because if the failure has originated in the user code, that user code might be in an invalid state. For example:
```
void notifyCheckpointComplete() {
a++;
maybeThrowException();
b++;
}
```
after throwing `a` and `b` will be inconsistent.
As we don't expose `notifyCheckpointSubsumed()` to the users, this argument doesn't hold here. So the only question is our internal data structures are expected to keep working just fine if an exception indeed happens in this call?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1168b48d412c2ac15da72d06bcdff90c2120f2c8 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725)
* 4575a58b2365b9ab070606108ef6ba1b4507dc24 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8ca91f13df7b3d17d2137e008e51acbcb4067c97 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r743366210
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
##########
@@ -339,78 +343,107 @@ public void checkpointState(
public void notifyCheckpointComplete(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
- Exception previousException = null;
- if (!isRunning.get()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for not-running task {}",
- checkpointId,
- taskName);
- } else if (operatorChain.isFinishedOnRestore()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for finished on restore task {}",
- checkpointId,
- taskName);
- } else {
- LOG.debug(
- "Notification of completed checkpoint {} for task {}", checkpointId, taskName);
- for (StreamOperatorWrapper<?, ?> operatorWrapper :
- operatorChain.getAllOperators(true)) {
- try {
- operatorWrapper.notifyCheckpointComplete(checkpointId);
- } catch (Exception e) {
- previousException = ExceptionUtils.firstOrSuppressed(e, previousException);
- }
- }
- }
- env.getTaskStateManager().notifyCheckpointComplete(checkpointId);
- ExceptionUtils.tryRethrowException(previousException);
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of complete checkpoint {} for not-running task {}",
+ "Ignoring notification of complete checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointComplete(checkpointId),
+ (tsm) -> tsm.notifyCheckpointComplete(checkpointId));
}
@Override
public void notifyCheckpointAborted(
long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of aborted checkpoint {} for not-running task {}",
+ "Ignoring notification of aborted checkpoint {} for finished on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ () -> {
+ boolean canceled = cancelAsyncCheckpointRunnable(checkpointId);
+
+ if (!canceled) {
+ if (checkpointId > lastCheckpointId) {
+ // only record checkpoints that have not triggered on task side.
+ abortedCheckpointIds.add(checkpointId);
+ }
+ }
+
+ channelStateWriter.abort(
+ checkpointId,
+ new CancellationException("checkpoint aborted via notification"),
+ false);
+ },
+ (opw) -> opw.getStreamOperator().notifyCheckpointAborted(checkpointId),
+ (tsm) -> tsm.notifyCheckpointAborted(checkpointId));
+ }
+
+ @Override
+ public void notifyCheckpointSubsumed(
+ long checkpointId, OperatorChain<?, ?> operatorChain, Supplier<Boolean> isRunning)
+ throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of subsumed checkpoint {} for not-running task {}",
+ "Ignoring notification of subsumed checkpoint {} for finished on restore task {}",
+ "Notification of subsumed checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointSubsumed(checkpointId),
+ null);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ OperatorChain<?, ?> operatorChain,
+ Supplier<Boolean> isRunning,
+ String logFormatIfNotRunning,
+ String logFormatIfFinishedOnRestore,
+ String logFormatIfNotificationStart,
+ @Nullable Runnable runnableBeforeNotifyOperator,
+ @Nonnull ThrowingConsumer<StreamOperatorWrapper<?, ?>, Exception> operatorConsumer,
+ @Nullable ThrowingConsumer<TaskStateManager, Exception> taskStateManagerConsumer)
+ throws Exception {
Review comment:
The refactor also take the `enum` of `NotifyCheckpointOperation` as `Task` did.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5817be659ecb4b6124f01415e552de29f538960b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r678109366
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1220,8 +1224,15 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
Preconditions.checkState(pendingCheckpoint.isDisposed() && completedCheckpoint != null);
try {
- completedCheckpointStore.addCheckpoint(
- completedCheckpoint, checkpointsCleaner, this::scheduleTriggerRequest);
+ CompletedCheckpoint lastSubsumed =
+ completedCheckpointStore.addCheckpointAndSubsumeOldestOne(
+ completedCheckpoint,
+ checkpointsCleaner,
+ this::scheduleTriggerRequest);
+ if (lastSubsumed != null && lastSubsumed.discardOnSubsume()) {
Review comment:
Why do we need `discardOnSubsume` check?
Shouldn't we notify about subsumption regardless of discarding (e.g. even if it's a savepoint)?
ditto: SubsumeHelper
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -229,6 +229,8 @@
private final ExecutionAttemptMappingProvider attemptMappingProvider;
+ @Nullable private CompletedCheckpoint lastSubsumedCheckpoint;
+
Review comment:
Why do we need this field?
ditto: `lastSubsumedCheckpointId`
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java
##########
@@ -67,6 +75,7 @@ public static void subsume(
}
// Don't break out from the loop to subsume intermediate savepoints
}
+ return lastSubsumedCheckpoint;
Review comment:
I think it would be more clear if the latest subsumed checkpoint is recorded by the store in it's subsume action (because of the same ambiguity about what is returned from this method).
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStore.java
##########
@@ -45,8 +47,12 @@
*
* <p>Only a bounded number of checkpoints is kept. When exceeding the maximum number of
* retained checkpoints, the oldest one will be discarded.
+ *
+ * @return the subsumed oldest completed checkpoint if possible, return null if no checkpoint
+ * needs to be discarded on subsume.
*/
- void addCheckpoint(
+ @Nullable
+ CompletedCheckpoint addCheckpointAndSubsumeOldestOne(
Review comment:
The new name doesn't seem to reflect what the method actually does.
It doesn't necessarily subsume **one** checkpoint (or **any** checkpoints at all).
The correct name would be `addAndMaybeSubsumeOldestAndReturnTheLatestSubsumedIfAny` - which clearly indicates to me that the new functionality should be in a separate method.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1243,6 +1250,29 @@ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint)
CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE,
exception);
}
+
+ try {
+ allPreviousCheckpoints.removeAll(completedCheckpointStore.getAllCheckpoints());
+ if (!allPreviousCheckpoints.isEmpty()) {
Review comment:
The new name doesn't seem to reflect what the method actually does.
It doesn't necessarily subsume **one** checkpoint (or **any** checkpoints at all).
The correct name would be `addAndMaybeSubsumeOldestAndReturnTheLatestSubsumedIfAny` - which clearly indicates to me that the new functionality should be in a separate method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r681416329
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java
##########
@@ -100,6 +102,16 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
}
}
+ public void notifyCheckpointSubsumed(long checkpointId) throws Exception {
+ if (!closed) {
+ if (wrapped instanceof AbstractStreamOperator) {
+ ((AbstractStreamOperator<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ } else if (wrapped instanceof AbstractStreamOperatorV2) {
+ ((AbstractStreamOperatorV2<?>) wrapped).notifyCheckpointSubsumed(checkpointId);
+ }
Review comment:
Got it, thanks for your explanation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16582: [FLINK-21504][checkpoint] Introduce notification of subsumed checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16582:
URL: https://github.com/apache/flink/pull/16582#issuecomment-885598535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20886",
"triggerID" : "935abc391163053cdc2d5fb35d5f321844f0e8b1",
"triggerType" : "PUSH"
}, {
"hash" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20979",
"triggerID" : "10466b2d0065049cc8d003cdfc1395785d279e23",
"triggerType" : "PUSH"
}, {
"hash" : "5817be659ecb4b6124f01415e552de29f538960b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21030",
"triggerID" : "5817be659ecb4b6124f01415e552de29f538960b",
"triggerType" : "PUSH"
}, {
"hash" : "36cfdc4abcad6028059c134f323e963d455921e8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21078",
"triggerID" : "36cfdc4abcad6028059c134f323e963d455921e8",
"triggerType" : "PUSH"
}, {
"hash" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21410",
"triggerID" : "dfdcff9fa15faf486afdb0ddd62c775f6dd61f3b",
"triggerType" : "PUSH"
}, {
"hash" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21474",
"triggerID" : "8a1fb9a7fc7ea43c249cbe7792102fe8ffe26321",
"triggerType" : "PUSH"
}, {
"hash" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=25991",
"triggerID" : "8ca91f13df7b3d17d2137e008e51acbcb4067c97",
"triggerType" : "PUSH"
}, {
"hash" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27725",
"triggerID" : "1168b48d412c2ac15da72d06bcdff90c2120f2c8",
"triggerType" : "PUSH"
}, {
"hash" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27764",
"triggerID" : "4575a58b2365b9ab070606108ef6ba1b4507dc24",
"triggerType" : "PUSH"
}, {
"hash" : "ab089745cc9841edb718c33a8ea3224775e38766",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27862",
"triggerID" : "ab089745cc9841edb718c33a8ea3224775e38766",
"triggerType" : "PUSH"
}, {
"hash" : "f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27900",
"triggerID" : "f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f6b8bfc5a5fabe54b6e40c8caf177d08e51c051e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=27900)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org