pnowojski commented on a change in pull request #13044:
URL: https://github.com/apache/flink/pull/13044#discussion_r484913033
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -528,15 +528,25 @@ private void
startTriggeringCheckpoint(CheckpointTriggerRequest request) {
request.getOnCompletionFuture()),
timer);
- final CompletableFuture<?> masterStatesComplete =
pendingCheckpointCompletableFuture
- .thenCompose(this::snapshotMasterState);
-
final CompletableFuture<?>
coordinatorCheckpointsComplete = pendingCheckpointCompletableFuture
.thenComposeAsync((pendingCheckpoint) ->
OperatorCoordinatorCheckpoints.triggerAndAcknowledgeAllCoordinatorCheckpointsWithCompletion(
coordinatorsToCheckpoint, pendingCheckpoint, timer),
timer);
+ // We have to take the snapshot of the master hooks
after the coordinator checkpoints has completed.
+ // This is to ensure the tasks are checkpointed after
the OperatorCoordinators in case
+ // ExternallyInducedSource is used.
Review comment:
nit: could you copy paste this comment to the commits description?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -955,7 +978,7 @@ public boolean
receiveAcknowledgeMessage(AcknowledgeCheckpoint message, String t
LOG.debug("Received acknowledge
message for checkpoint {} from task {} of job {} at {}.",
checkpointId,
message.getTaskExecutionId(), message.getJob(), taskManagerLocationInfo);
- if
(checkpoint.areTasksFullyAcknowledged()) {
+ if
(checkpoint.isFullyAcknowledged()) {
Review comment:
Is this change part of the same bug fix as the changes above?
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
##########
@@ -2436,6 +2442,126 @@ public Integer deserialize(int version, byte[]
serialized) throws IOException {
coord.shutdown(JobStatus.FINISHED);
}
+ @Test
+ public void testCompleteCheckpointFailureWithExternallyInducedSource()
throws Exception {
Review comment:
why is this added in a separate commit?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
