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 2020/03/09 04:05:12 UTC
[GitHub] [flink] ifndef-SleePy opened a new pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
ifndef-SleePy opened a new pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347
## What is the purpose of the change
* This is the last part of refactoring threading model of `CheckpointCoordinator`
* The threading model of `CheckpointCoordinator` would be simplified a lot
* All the non-IO operations are executed in a single-threaded way
* A lot of competitions inside or outside `CheckpointCoordinator` could be avoided. So it's easier for `CheckpointCoordinator` to cooperate with other components
## Brief change log
* Split the ACK and declined message handling into two ways
* Handle non-IO operations in timer thread
* Handle IO operations in IO thread
* This is a preparation of introducing main thread executor in `CheckpointCoordinator`. After this, there would be non-IO operations executed in IO thread
* Introduce main thread executor in `CheckpointCoordinator`
* It's used to execute all non-IO operations instead of the timer thread now
* The timer thread would be kept for now. It's only used to schedule the periodic triggering, because currently main thread executor does not support periodic scheduling yet.
* The coordinator-wide lock of `CheckpointCoordinator` and lock of `PendingCheckpoint` could be avoided because there is no thread competition after introducing main thread executor
## Verifying this change
* This change is already covered by existing tests, and some test cases are added at the same time
## 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/Mesos, 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-602485849
Hi @pnowojski , I have addressed all comments and done the squashing. Could you please take a look?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r392859883
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -311,25 +315,32 @@ public CheckpointException getFailureCause() {
try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) {
Checkpoints.storeCheckpointMetadata(savepoint, out);
finalizedLocation = out.closeAndFinalizeCheckpoint();
+ }
+ CompletedCheckpoint completed = new CompletedCheckpoint(
+ jobId,
+ checkpointId,
+ checkpointTimestamp,
+ System.currentTimeMillis(),
+ operatorStates,
+ masterStates,
+ props,
+ finalizedLocation);
+
+ try {
+ completedCheckpointStore.addCheckpoint(completed);
+ } catch (Throwable t) {
+ completed.discardOnFailedStoring();
Review comment:
However I do find a problem inspired by this comment, nice job!
There might be a cancellation or shutting down during the finalization. It's not well handled yet, I would update the PR later to cover this scenario.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r390894422
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1018,61 +1019,64 @@ else if (checkpoint != null) {
* <p>Important: This method should only be called in the checkpoint lock scope.
*
* @param pendingCheckpoint to complete
- * @throws CheckpointException if the completion failed
*/
- private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint) throws CheckpointException {
- final long checkpointId = pendingCheckpoint.getCheckpointId();
- final CompletedCheckpoint completedCheckpoint;
-
+ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint) {
// As a first step to complete the checkpoint, we register its state with the registry
Map<OperatorID, OperatorState> operatorStates = pendingCheckpoint.getOperatorStates();
sharedStateRegistry.registerAll(operatorStates.values());
- try {
- try {
- completedCheckpoint = pendingCheckpoint.finalizeCheckpoint();
- failureManager.handleCheckpointSuccess(pendingCheckpoint.getCheckpointId());
- }
- catch (Exception e1) {
- // abort the current pending checkpoint if we fails to finalize the pending checkpoint.
- if (!pendingCheckpoint.isDiscarded()) {
- abortPendingCheckpoint(
- pendingCheckpoint,
- new CheckpointException(
- CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, e1));
+ final CompletableFuture<CompletedCheckpoint> completedCheckpointFuture = pendingCheckpoint.finalizeCheckpoint();
+ completedCheckpointFuture.thenApplyAsync((completedCheckpoint) -> {
+ synchronized (lock) {
+ if (shutdown) {
+ return null;
+ }
+ // the pending checkpoint must be discarded after the finalization
+ Preconditions.checkState(pendingCheckpoint.isDiscarded() && completedCheckpoint != null);
+ try {
+ completedCheckpointStore.addCheckpoint(completedCheckpoint);
+ return completedCheckpoint;
+ } catch (Throwable t) {
+ try {
+ completedCheckpoint.discardOnFailedStoring();
+ } catch (Exception e) {
+ LOG.warn("Could not properly discard completed checkpoint {}.", completedCheckpoint.getCheckpointID(), e);
+ }
+ throw new CompletionException(t);
}
-
- throw new CheckpointException("Could not finalize the pending checkpoint " + checkpointId + '.',
- CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, e1);
}
-
- // the pending checkpoint must be discarded after the finalization
- Preconditions.checkState(pendingCheckpoint.isDiscarded() && completedCheckpoint != null);
-
- try {
- completedCheckpointStore.addCheckpoint(completedCheckpoint);
- } catch (Exception exception) {
- // we failed to store the completed checkpoint. Let's clean up
- executor.execute(new Runnable() {
- @Override
- public void run() {
- try {
- completedCheckpoint.discardOnFailedStoring();
- } catch (Throwable t) {
- LOG.warn("Could not properly discard completed checkpoint {}.", completedCheckpoint.getCheckpointID(), t);
- }
+ }, executor)
+ .whenCompleteAsync((completedCheckpoint, throwable) -> {
+ synchronized (lock) {
+ if (shutdown) {
+ return;
+ }
+ if (throwable != null) {
+ if (!pendingCheckpoint.isDiscarded()) {
+ abortPendingCheckpoint(
+ pendingCheckpoint,
+ new CheckpointException(
+ CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, throwable));
}
- });
-
- throw new CheckpointException("Could not complete the pending checkpoint " + checkpointId + '.',
- CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, exception);
+ } else {
+ onCheckpointSuccess(completedCheckpoint);
+ }
}
- } finally {
- pendingCheckpoints.remove(checkpointId);
+ }, timer);
+ }
Review comment:
I think we are missing some comments why is something being executed on the main thread or io thread executor. For example this chain in CheckpointCoordinator#completePendingCheckpoint:
1. finalizeCheckpoint goes first on io executor
2. then finalizeCheckpoint asynchronously goes back to the main thread
3. then completePendingCheckpoint goes again to io executor
4. and finally completePendingCheckpoint goes back to main thread executor asynchronously
First thing is those missing comments, why something is executed in one executor not the other one. Secondly, does it have to be this back and forth? Could we somehow simplify the code with simpler control flow `main thread -> io thread -> main thread`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395000464
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -304,35 +307,44 @@ public CheckpointException getFailureCause() {
"Pending checkpoint has not been fully acknowledged yet"));
}
+ // now we stop the canceller before finalization
+ // it simplifies the concurrent conflict issue here
+ cancelCanceller();
+
// make sure we fulfill the promise with an exception if something fails
final CompletableFuture<CompletedCheckpoint> finalizedLocationFuture =
CompletableFuture.supplyAsync(() -> {
try {
- // write out the metadata
- final CheckpointMetadata savepoint = new CheckpointMetadata(checkpointId, operatorStates.values(), masterStates);
- final CompletedCheckpointStorageLocation finalizedLocation;
+ synchronized (operationLock) {
+ if (isDiscarded()) {
+ throw new IllegalStateException("The checkpoint has been discarded");
+ }
Review comment:
nit: `checkState()`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-599714919
Hi @pnowojski , I've just updated the PR to resolve the concurrent conflict issue while finalizing checkpoint. Could you please take a look?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
}, {
"hash" : "655adff5f2d9298b34b167b5a355f4b536613246",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153558707",
"triggerID" : "655adff5f2d9298b34b167b5a355f4b536613246",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 655adff5f2d9298b34b167b5a355f4b536613246 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153558707)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395010728
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -342,6 +354,12 @@ public CheckpointException getFailureCause() {
return finalizedLocationFuture.thenApplyAsync((completed) -> {
+ if (isDiscarded()) {
+ // since canceller has been already cancelled, the coordinator must be shut down
+ // all the resources should be released properly when it's shutting down the coordinator
+ throw new IllegalStateException("The checkpoint has been discarded");
+ }
Review comment:
`checkState`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cf8adb3d042bb89172fb45b68b6448713b4045ae Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153468080)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a04e4412520adcc0f51cb9ce2b53c8ea70798bcc Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/152868791) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395429228
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -111,6 +108,14 @@
/** The executor for potentially blocking I/O operations, like state disposal. */
private final Executor executor;
+ /** The executor for non-blocking operations. */
+ private final Executor mainThreadExecutor;
+
+ private final CompletedCheckpointStore completedCheckpointStore;
+
+ /** The lock for avoiding conflict between I/O operations. */
+ private final Object operationLock = new Object();
Review comment:
Yes, there is a small possibility that the `CheckpointCoordinator` is shut down when a `PendingCheckpoint` is doing finalization. There could be some concurrent conflicts on `operatorStates` and `targetLocation`. It might be not a big deal because it would be shut down anyway. The finalization probably could not finish because the IO executor would be also shut down. However it's not so elegant to leave the concurrent issue to the `CheckpointStorageLocation` and `OperatorState`. And it's a bit heavy to make all of these implementations thread-safe to avoid the small possibility issue. So here I think introducing a lock outside is better.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395429422
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -91,9 +92,6 @@
// ------------------------------------------------------------------------
- /** Coordinator-wide lock to safeguard the checkpoint updates. */
- private final Object lock = new Object();
Review comment:
Sure, nice point!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r392801274
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -403,7 +403,10 @@ public void shutdown(JobStatus jobStatus) throws Exception {
// clear queued requests and in-flight checkpoints
abortPendingAndQueuedCheckpoints(reason);
- completedCheckpointStore.shutdown(jobStatus);
+ // there might be a race condition with IO threads on completedCheckpointStore
+ synchronized (completedCheckpointStore) {
Review comment:
Sounds reasonable.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "CANCELED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d5078d3b1229c6391cdfe2394538ef810e9e01d4 Travis: [CANCELED](https://travis-ci.com/flink-ci/flink/builds/152384638) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068)
* 3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/152395537) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
}, {
"hash" : "655adff5f2d9298b34b167b5a355f4b536613246",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153558707",
"triggerID" : "655adff5f2d9298b34b167b5a355f4b536613246",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cf8adb3d042bb89172fb45b68b6448713b4045ae Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153468080)
* 655adff5f2d9298b34b167b5a355f4b536613246 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153558707)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r391383507
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1018,61 +1019,64 @@ else if (checkpoint != null) {
* <p>Important: This method should only be called in the checkpoint lock scope.
*
* @param pendingCheckpoint to complete
- * @throws CheckpointException if the completion failed
*/
- private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint) throws CheckpointException {
- final long checkpointId = pendingCheckpoint.getCheckpointId();
- final CompletedCheckpoint completedCheckpoint;
-
+ private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint) {
// As a first step to complete the checkpoint, we register its state with the registry
Map<OperatorID, OperatorState> operatorStates = pendingCheckpoint.getOperatorStates();
sharedStateRegistry.registerAll(operatorStates.values());
- try {
- try {
- completedCheckpoint = pendingCheckpoint.finalizeCheckpoint();
- failureManager.handleCheckpointSuccess(pendingCheckpoint.getCheckpointId());
- }
- catch (Exception e1) {
- // abort the current pending checkpoint if we fails to finalize the pending checkpoint.
- if (!pendingCheckpoint.isDiscarded()) {
- abortPendingCheckpoint(
- pendingCheckpoint,
- new CheckpointException(
- CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, e1));
+ final CompletableFuture<CompletedCheckpoint> completedCheckpointFuture = pendingCheckpoint.finalizeCheckpoint();
+ completedCheckpointFuture.thenApplyAsync((completedCheckpoint) -> {
+ synchronized (lock) {
+ if (shutdown) {
+ return null;
+ }
+ // the pending checkpoint must be discarded after the finalization
+ Preconditions.checkState(pendingCheckpoint.isDiscarded() && completedCheckpoint != null);
+ try {
+ completedCheckpointStore.addCheckpoint(completedCheckpoint);
+ return completedCheckpoint;
+ } catch (Throwable t) {
+ try {
+ completedCheckpoint.discardOnFailedStoring();
+ } catch (Exception e) {
+ LOG.warn("Could not properly discard completed checkpoint {}.", completedCheckpoint.getCheckpointID(), e);
+ }
+ throw new CompletionException(t);
}
-
- throw new CheckpointException("Could not finalize the pending checkpoint " + checkpointId + '.',
- CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, e1);
}
-
- // the pending checkpoint must be discarded after the finalization
- Preconditions.checkState(pendingCheckpoint.isDiscarded() && completedCheckpoint != null);
-
- try {
- completedCheckpointStore.addCheckpoint(completedCheckpoint);
- } catch (Exception exception) {
- // we failed to store the completed checkpoint. Let's clean up
- executor.execute(new Runnable() {
- @Override
- public void run() {
- try {
- completedCheckpoint.discardOnFailedStoring();
- } catch (Throwable t) {
- LOG.warn("Could not properly discard completed checkpoint {}.", completedCheckpoint.getCheckpointID(), t);
- }
+ }, executor)
+ .whenCompleteAsync((completedCheckpoint, throwable) -> {
+ synchronized (lock) {
+ if (shutdown) {
+ return;
+ }
+ if (throwable != null) {
+ if (!pendingCheckpoint.isDiscarded()) {
+ abortPendingCheckpoint(
+ pendingCheckpoint,
+ new CheckpointException(
+ CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, throwable));
}
- });
-
- throw new CheckpointException("Could not complete the pending checkpoint " + checkpointId + '.',
- CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, exception);
+ } else {
+ onCheckpointSuccess(completedCheckpoint);
+ }
}
- } finally {
- pendingCheckpoints.remove(checkpointId);
+ }, timer);
+ }
Review comment:
It's indeed a bit complicated. If we combine the meta data persistence and completed checkpoint storing in one operation (in IO thread), the logic here would be much easier to understand.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r392856449
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -311,25 +315,32 @@ public CheckpointException getFailureCause() {
try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) {
Checkpoints.storeCheckpointMetadata(savepoint, out);
finalizedLocation = out.closeAndFinalizeCheckpoint();
+ }
+ CompletedCheckpoint completed = new CompletedCheckpoint(
+ jobId,
+ checkpointId,
+ checkpointTimestamp,
+ System.currentTimeMillis(),
+ operatorStates,
+ masterStates,
+ props,
+ finalizedLocation);
+
+ try {
+ completedCheckpointStore.addCheckpoint(completed);
+ } catch (Throwable t) {
+ completed.discardOnFailedStoring();
Review comment:
Good question!
Actually `completedCheckpointStore.addCheckpoint` should be called before `finalizedLocationFuture.thenApplyAsync((completed)`. The `finalizedLocationFuture.thenApplyAsync((completed)` does things like completing `onCompletionPromise`, reporting completed statistics, disposing the pending checkpoint. However if `completedCheckpointStore.addCheckpoint` fails afterwards, does this checkpoint succeeds? I don't think so. But `onCompletionPromise` has been completed in this scenario. It's inconsistent here.
So the right way here is calling `completedCheckpointStore.addCheckpoint` first, then completing `onCompletionPromise`.
I was planning to do this as a follow-up issue. However since we have decided to combine the finalization and adding into `completedCheckpointStore` to simplify the operations between IO threads and main thread, I think it's a good opportunity to do 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395012640
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -91,9 +92,6 @@
// ------------------------------------------------------------------------
- /** Coordinator-wide lock to safeguard the checkpoint updates. */
- private final Object lock = new Object();
Review comment:
nit: some comments like L173 and L176 I think need to be updated, as there is no "synchronized" scope anymore but just single threaded?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski merged pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski merged pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
}, {
"hash" : "655adff5f2d9298b34b167b5a355f4b536613246",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153558707",
"triggerID" : "655adff5f2d9298b34b167b5a355f4b536613246",
"triggerType" : "PUSH"
}, {
"hash" : "9f59db6be464318410e6ce165cb599233980b534",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9f59db6be464318410e6ce165cb599233980b534",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 655adff5f2d9298b34b167b5a355f4b536613246 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153558707)
* 9f59db6be464318410e6ce165cb599233980b534 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-597988749
Hey @pnowojski , thanks for reviewing. The e2e testing case got passed this time, I think it's an unstable case. I will file a ticket for this. Thanks for reminding.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596411528
@flinkbot run azure
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r391494363
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -311,25 +315,32 @@ public CheckpointException getFailureCause() {
try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) {
Checkpoints.storeCheckpointMetadata(savepoint, out);
finalizedLocation = out.closeAndFinalizeCheckpoint();
+ }
+ CompletedCheckpoint completed = new CompletedCheckpoint(
+ jobId,
+ checkpointId,
+ checkpointTimestamp,
+ System.currentTimeMillis(),
+ operatorStates,
+ masterStates,
+ props,
+ finalizedLocation);
+
+ try {
+ completedCheckpointStore.addCheckpoint(completed);
+ } catch (Throwable t) {
+ completed.discardOnFailedStoring();
Review comment:
could there be some side effects on changing order of the operations? Now those lines are executed before `inalizedLocationFuture.thenApplyAsync((completed) ...` on L343.
What would happen if there is some exception etc?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395020929
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -311,25 +315,32 @@ public CheckpointException getFailureCause() {
try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) {
Checkpoints.storeCheckpointMetadata(savepoint, out);
finalizedLocation = out.closeAndFinalizeCheckpoint();
+ }
+ CompletedCheckpoint completed = new CompletedCheckpoint(
+ jobId,
+ checkpointId,
+ checkpointTimestamp,
+ System.currentTimeMillis(),
+ operatorStates,
+ masterStates,
+ props,
+ finalizedLocation);
+
+ try {
+ completedCheckpointStore.addCheckpoint(completed);
+ } catch (Throwable t) {
+ completed.discardOnFailedStoring();
Review comment:
I assume that one (or both) of the tests that you have added in the last commit were failing before fixing the issue?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d5078d3b1229c6391cdfe2394538ef810e9e01d4 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r391492234
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -403,7 +403,10 @@ public void shutdown(JobStatus jobStatus) throws Exception {
// clear queued requests and in-flight checkpoints
abortPendingAndQueuedCheckpoints(reason);
- completedCheckpointStore.shutdown(jobStatus);
+ // there might be a race condition with IO threads on completedCheckpointStore
+ synchronized (completedCheckpointStore) {
Review comment:
If we are synchronizing all of the access to the `completedCheckpointStore` externally, wouldn't it be better to make `CompletedCheckpointStore` `@ThreadSafe` and synchronise it inside?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
}, {
"hash" : "655adff5f2d9298b34b167b5a355f4b536613246",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153558707",
"triggerID" : "655adff5f2d9298b34b167b5a355f4b536613246",
"triggerType" : "PUSH"
}, {
"hash" : "9f59db6be464318410e6ce165cb599233980b534",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9f59db6be464318410e6ce165cb599233980b534",
"triggerType" : "PUSH"
}, {
"hash" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/154559672",
"triggerID" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"triggerType" : "PUSH"
}, {
"hash" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6514",
"triggerID" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9f59db6be464318410e6ce165cb599233980b534 UNKNOWN
* c48d0125ddfda31c4d4c9aede97af4e3c818a3d9 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/154559672) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6514)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596362868
@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395004574
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -111,6 +108,14 @@
/** The executor for potentially blocking I/O operations, like state disposal. */
private final Executor executor;
+ /** The executor for non-blocking operations. */
+ private final Executor mainThreadExecutor;
+
+ private final CompletedCheckpointStore completedCheckpointStore;
+
+ /** The lock for avoiding conflict between I/O operations. */
+ private final Object operationLock = new Object();
Review comment:
This is reverting a lock here. The issue is that we were able to make `CheckpointCoordinator` code that's executed in the `mainThreadExecutor` lock free, as `mainThreadExecutor` is single threaded. But we still need lock for IO operations executed in the io executor?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596363068
@flinkbot run azure
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r390285228
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -588,8 +588,8 @@ private void startTriggeringCheckpoint(
final CompletableFuture<?> coordinatorCheckpointsComplete = pendingCheckpointCompletableFuture
.thenComposeAsync((pendingCheckpoint) ->
OperatorCoordinatorCheckpoints.triggerAndAcknowledgeAllCoordinatorCheckpointsWithCompletion(
- coordinatorsToCheckpoint, pendingCheckpoint, timer),
- timer);
+ coordinatorsToCheckpoint, pendingCheckpoint, mainThreadExecutor),
Review comment:
Shouldn't this commit:
> [hotfix] Fix broken DefaultSchedulerTest by satisfying the main thread requirement of CheckpointCoordinator
be squashed with the preceding one?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d5078d3b1229c6391cdfe2394538ef810e9e01d4 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/152384638) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068)
* 3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy removed a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy removed a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596362868
@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596363028
@flinkbot run travis
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r396315368
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -311,25 +315,32 @@ public CheckpointException getFailureCause() {
try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) {
Checkpoints.storeCheckpointMetadata(savepoint, out);
finalizedLocation = out.closeAndFinalizeCheckpoint();
+ }
+ CompletedCheckpoint completed = new CompletedCheckpoint(
+ jobId,
+ checkpointId,
+ checkpointTimestamp,
+ System.currentTimeMillis(),
+ operatorStates,
+ masterStates,
+ props,
+ finalizedLocation);
+
+ try {
+ completedCheckpointStore.addCheckpoint(completed);
+ } catch (Throwable t) {
+ completed.discardOnFailedStoring();
Review comment:
It should be.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/152395537) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596322115
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 d5078d3b1229c6391cdfe2394538ef810e9e01d4 (Mon Mar 09 04:08:26 UTC 2020)
**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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a04e4412520adcc0f51cb9ce2b53c8ea70798bcc Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/152868791) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r391493247
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -403,7 +403,10 @@ public void shutdown(JobStatus jobStatus) throws Exception {
// clear queued requests and in-flight checkpoints
abortPendingAndQueuedCheckpoints(reason);
- completedCheckpointStore.shutdown(jobStatus);
+ // there might be a race condition with IO threads on completedCheckpointStore
Review comment:
The race condition is between io threads and just the shut down? And previously was it working becase `CompletedCheckpointStore` access were synchronized on the checkopint coordinator's lock?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d5078d3b1229c6391cdfe2394538ef810e9e01d4 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/152384638) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
}, {
"hash" : "655adff5f2d9298b34b167b5a355f4b536613246",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153558707",
"triggerID" : "655adff5f2d9298b34b167b5a355f4b536613246",
"triggerType" : "PUSH"
}, {
"hash" : "9f59db6be464318410e6ce165cb599233980b534",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9f59db6be464318410e6ce165cb599233980b534",
"triggerType" : "PUSH"
}, {
"hash" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/154559672",
"triggerID" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"triggerType" : "PUSH"
}, {
"hash" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6514",
"triggerID" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9f59db6be464318410e6ce165cb599233980b534 UNKNOWN
* c48d0125ddfda31c4d4c9aede97af4e3c818a3d9 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/154559672) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6514)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395431078
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -151,6 +149,9 @@
* It must be single-threaded. Eventually it will be replaced by main thread executor. */
private final ScheduledExecutor timer;
+ @Nullable
+ private ComponentMainThreadExecutor mainThreadExecutor;
Review comment:
Yes, it could not be a `final` variable based on current implementation of `ExecutionGraph`. I'm just not a fan of the trick of `DummyComponentMainThreadExecutor`. But I agree with you that it's not a bad idea to provide a consistent way with `ExecutionGraph`. Let me think of it a bit more. Maybe we could find a better solution.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r394983835
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStore.java
##########
@@ -28,6 +28,7 @@
/**
* A bounded LIFO-queue of {@link CompletedCheckpoint} instances.
+ * Note that it might be visited by multiple threads. So implementation should keep it thread-safe.
Review comment:
```
@ThreadSafe
```
on the interface?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r392856449
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
##########
@@ -311,25 +315,32 @@ public CheckpointException getFailureCause() {
try (CheckpointMetadataOutputStream out = targetLocation.createMetadataOutputStream()) {
Checkpoints.storeCheckpointMetadata(savepoint, out);
finalizedLocation = out.closeAndFinalizeCheckpoint();
+ }
+ CompletedCheckpoint completed = new CompletedCheckpoint(
+ jobId,
+ checkpointId,
+ checkpointTimestamp,
+ System.currentTimeMillis(),
+ operatorStates,
+ masterStates,
+ props,
+ finalizedLocation);
+
+ try {
+ completedCheckpointStore.addCheckpoint(completed);
+ } catch (Throwable t) {
+ completed.discardOnFailedStoring();
Review comment:
Good question!
Actually `completedCheckpointStore.addCheckpoint` should be called before `finalizedLocationFuture.thenApplyAsync((completed)`. The `finalizedLocationFuture.thenApplyAsync((completed)` does things like completing `onCompletionPromise`, reporting completed statistics, disposing the pending checkpoint. However if `completedCheckpointStore.addCheckpoint` fails afterwards, does this checkpoint succeeds? I don't think so. But `onCompletionPromise` has been completed in this scenario. It's inconsistent here.
So the right way here is calling `completedCheckpointStore.addCheckpoint` first, then completing `onCompletionPromise`.I was planning to do this as a follow-up issue. However since we have decided to combine the finalization and adding into `completedCheckpointStore`,
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d5078d3b1229c6391cdfe2394538ef810e9e01d4 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/152384638) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
}, {
"hash" : "655adff5f2d9298b34b167b5a355f4b536613246",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "655adff5f2d9298b34b167b5a355f4b536613246",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cf8adb3d042bb89172fb45b68b6448713b4045ae Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153468080)
* 655adff5f2d9298b34b167b5a355f4b536613246 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
}, {
"hash" : "655adff5f2d9298b34b167b5a355f4b536613246",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153558707",
"triggerID" : "655adff5f2d9298b34b167b5a355f4b536613246",
"triggerType" : "PUSH"
}, {
"hash" : "9f59db6be464318410e6ce165cb599233980b534",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9f59db6be464318410e6ce165cb599233980b534",
"triggerType" : "PUSH"
}, {
"hash" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/154559672",
"triggerID" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"triggerType" : "PUSH"
}, {
"hash" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6514",
"triggerID" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 655adff5f2d9298b34b167b5a355f4b536613246 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153558707)
* 9f59db6be464318410e6ce165cb599233980b534 UNKNOWN
* c48d0125ddfda31c4d4c9aede97af4e3c818a3d9 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/154559672) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6514)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r396319772
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -151,6 +149,9 @@
* It must be single-threaded. Eventually it will be replaced by main thread executor. */
private final ScheduledExecutor timer;
+ @Nullable
+ private ComponentMainThreadExecutor mainThreadExecutor;
Review comment:
Sadly I didn't find an easy way. It involves too much for this. For now I use the `DummyComponentMainThreadExecutor` and mark it `@Nonnull` as you advised. But I think it's worth a separate ticket. I put it on my follow-up list.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r392803884
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -403,7 +403,10 @@ public void shutdown(JobStatus jobStatus) throws Exception {
// clear queued requests and in-flight checkpoints
abortPendingAndQueuedCheckpoints(reason);
- completedCheckpointStore.shutdown(jobStatus);
+ // there might be a race condition with IO threads on completedCheckpointStore
Review comment:
The race condition happens between `completedCheckpointStore.shutdown` and `completedCheckpointStore.addCheckpoint`.
> And previously was it working becase CompletedCheckpointStore access were synchronized on the checkopint coordinator's lock?
Yes, exactly.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152395537) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073)
* a04e4412520adcc0f51cb9ce2b53c8ea70798bcc Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/152868791) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152395537) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073)
* a04e4412520adcc0f51cb9ce2b53c8ea70798bcc 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a04e4412520adcc0f51cb9ce2b53c8ea70798bcc Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/152868791) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197)
* cf8adb3d042bb89172fb45b68b6448713b4045ae Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153468080)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] ifndef-SleePy commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
ifndef-SleePy commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r391381769
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -588,8 +588,8 @@ private void startTriggeringCheckpoint(
final CompletableFuture<?> coordinatorCheckpointsComplete = pendingCheckpointCompletableFuture
.thenComposeAsync((pendingCheckpoint) ->
OperatorCoordinatorCheckpoints.triggerAndAcknowledgeAllCoordinatorCheckpointsWithCompletion(
- coordinatorsToCheckpoint, pendingCheckpoint, timer),
- timer);
+ coordinatorsToCheckpoint, pendingCheckpoint, mainThreadExecutor),
Review comment:
Oops, I just realize this commit includes some codes belongs the preceding commit. It's must caused by the conflict resolving. And yes, you are right. This commit could be squashed with the preceding one. I will do the squashing when all comments are addressed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153468080",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
}, {
"hash" : "655adff5f2d9298b34b167b5a355f4b536613246",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153558707",
"triggerID" : "655adff5f2d9298b34b167b5a355f4b536613246",
"triggerType" : "PUSH"
}, {
"hash" : "9f59db6be464318410e6ce165cb599233980b534",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9f59db6be464318410e6ce165cb599233980b534",
"triggerType" : "PUSH"
}, {
"hash" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c48d0125ddfda31c4d4c9aede97af4e3c818a3d9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 655adff5f2d9298b34b167b5a355f4b536613246 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153558707)
* 9f59db6be464318410e6ce165cb599233980b534 UNKNOWN
* c48d0125ddfda31c4d4c9aede97af4e3c818a3d9 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152395537) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "596363028",
"triggerType" : "MANUAL"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6073",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152395537",
"triggerID" : "3bac3485e8fa916db3c1ec7dffcf5b80dbaef4de",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152868791",
"triggerID" : "a04e4412520adcc0f51cb9ce2b53c8ea70798bcc",
"triggerType" : "PUSH"
}, {
"hash" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "cf8adb3d042bb89172fb45b68b6448713b4045ae",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a04e4412520adcc0f51cb9ce2b53c8ea70798bcc Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/152868791) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6197)
* cf8adb3d042bb89172fb45b68b6448713b4045ae 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#issuecomment-596323413
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
}, {
"hash" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384638",
"triggerID" : "d5078d3b1229c6391cdfe2394538ef810e9e01d4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d5078d3b1229c6391cdfe2394538ef810e9e01d4 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/152384638) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6068)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11347:
[FLINK-14971][checkpointing] Make all the non-IO operations in
CheckpointCoordinator single-threaded
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11347: [FLINK-14971][checkpointing] Make all the non-IO operations in CheckpointCoordinator single-threaded
URL: https://github.com/apache/flink/pull/11347#discussion_r395017182
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -151,6 +149,9 @@
* It must be single-threaded. Eventually it will be replaced by main thread executor. */
private final ScheduledExecutor timer;
+ @Nullable
+ private ComponentMainThreadExecutor mainThreadExecutor;
Review comment:
It's not nice that this is not a `final` field, but I guess this would need large scope of changes, as there is the same issue in `ExecutionGraph`.
However, why does it have to be `@Nullable`? `ExecutionGraph#jobMasterMainThreadExecutor` uses interesting trick with `DummyComponentMainThreadExecutor`. I'm not sure if doing `instanceof` checks is better or not, but maybe for the sake of consistency we should do the same 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services