You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/01/25 16:57:09 UTC
[flink] 08/11: [FLINK-19462][checkpointing] Report aborted
checkpoint stats
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 621ea85ed8193cae34c24f94f3f8d9ccc5faad40
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Tue Jan 12 20:36:59 2021 +0100
[FLINK-19462][checkpointing] Report aborted checkpoint stats
This change introduces a new RPC from TM to JM.
Existing one can't be used because it:
a) confirms the checkpoint
b) requires task state snapshot
The call is issued after cancelling task state-persisting
futures upon receiving abortion notification. This way
more precise metrics are available (compared to reporting
from AsyncCheckpointRunnable after cancellation).
---
.../api/runtime/SavepointTaskStateManager.java | 4 +++
.../subtask/job-checkpoints-subtask.component.html | 2 +-
.../checkpoint/CheckpointCoordinatorGateway.java | 6 +++++
.../runtime/checkpoint/CheckpointStatsTracker.java | 3 ++-
.../runtime/checkpoint/PendingCheckpoint.java | 3 ++-
.../runtime/checkpoint/PendingCheckpointStats.java | 6 +++--
.../runtime/checkpoint/SubtaskStateStats.java | 11 +++++++-
.../flink/runtime/checkpoint/TaskStateStats.java | 10 ++++---
.../apache/flink/runtime/jobmaster/JobMaster.java | 11 ++++++++
.../TaskCheckpointStatisticDetailsHandler.java | 3 ++-
.../checkpoints/SubtaskCheckpointStatistics.java | 19 ++++++++++---
.../flink/runtime/scheduler/SchedulerBase.java | 28 +++++++++++++++++++
.../flink/runtime/scheduler/SchedulerNG.java | 6 +++++
.../flink/runtime/state/TaskStateManager.java | 9 +++++++
.../flink/runtime/state/TaskStateManagerImpl.java | 8 ++++++
.../taskexecutor/rpc/RpcCheckpointResponder.java | 10 +++++++
.../runtime/taskmanager/CheckpointResponder.java | 16 +++++++++++
.../checkpoint/CheckpointCoordinatorTest.java | 31 +++++++++++++++++++---
.../checkpoint/CheckpointStatsTrackerTest.java | 5 ++--
.../checkpoint/CompletedCheckpointTest.java | 2 +-
.../checkpoint/PendingCheckpointStatsTest.java | 3 ++-
.../runtime/checkpoint/SubtaskStateStatsTest.java | 3 ++-
.../runtime/checkpoint/TaskStateStatsTest.java | 5 ++--
.../jobmaster/utils/TestingJobMasterGateway.java | 7 +++++
...CheckpointStatisticsWithSubtaskDetailsTest.java | 3 ++-
.../runtime/scheduler/TestingSchedulerNG.java | 7 +++++
.../flink/runtime/state/TestTaskStateManager.java | 4 +++
.../taskmanager/NoOpCheckpointResponder.java | 7 +++++
.../taskmanager/TestCheckpointResponder.java | 7 +++++
.../streaming/state/RocksDBAsyncSnapshotTest.java | 7 +++++
.../runtime/tasks/AsyncCheckpointRunnable.java | 16 ++++++++++-
.../tasks/TaskCheckpointingBehaviourTest.java | 7 +++++
32 files changed, 243 insertions(+), 26 deletions(-)
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointTaskStateManager.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointTaskStateManager.java
index 5237590..95355a5 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointTaskStateManager.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointTaskStateManager.java
@@ -54,6 +54,10 @@ final class SavepointTaskStateManager implements TaskStateManager {
@Nullable TaskStateSnapshot acknowledgedState,
@Nullable TaskStateSnapshot localState) {}
+ @Override
+ public void reportIncompleteTaskStateSnapshots(
+ CheckpointMetaData checkpointMetaData, CheckpointMetrics checkpointMetrics) {}
+
@Nonnull
@Override
public PrioritizedOperatorSubtaskState prioritizedOperatorState(OperatorID operatorID) {
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/subtask/job-checkpoints-subtask.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/subtask/job-checkpoints-subtask.component.html
index 79bff5f..9510eca 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/subtask/job-checkpoints-subtask.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/subtask/job-checkpoints-subtask.component.html
@@ -99,7 +99,7 @@
<td>{{ subTask['index'] }}</td>
<ng-container *ngIf="subTask['status'] == 'completed'">
<td >{{ subTask['ack_timestamp'] | date:'yyyy-MM-dd HH:mm:ss' }}</td>
- <td>{{ subTask['end_to_end_duration'] | humanizeDuration}}</td>
+ <td>{{ subTask['end_to_end_duration'] | humanizeDuration}} <span *ngIf="subTask['aborted']">(aborted)</span></td>
<td>{{ subTask['state_size'] | humanizeBytes }}</td>
<td>{{ subTask['checkpoint']['sync'] | humanizeDuration}}</td>
<td>{{ subTask['checkpoint']['async'] | humanizeDuration}}</td>
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorGateway.java
index 532b69c..cdc7b00 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorGateway.java
@@ -34,4 +34,10 @@ public interface CheckpointCoordinatorGateway extends RpcGateway {
final TaskStateSnapshot subtaskState);
void declineCheckpoint(DeclineCheckpoint declineCheckpoint);
+
+ void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
index 04cfcf2..e8883dd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
@@ -274,7 +274,8 @@ public class CheckpointStatsTracker {
metrics.getBytesPersistedDuringAlignment(),
metrics.getAlignmentDurationNanos() / 1_000_000,
metrics.getCheckpointStartDelayNanos() / 1_000_000,
- metrics.getUnalignedCheckpoint()));
+ metrics.getUnalignedCheckpoint(),
+ false));
dirty = true;
}
} finally {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
index 33dfd21..a2e1784 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
@@ -439,7 +439,8 @@ public class PendingCheckpoint implements Checkpoint {
metrics.getBytesPersistedDuringAlignment(),
alignmentDurationMillis,
checkpointStartDelayMillis,
- metrics.getUnalignedCheckpoint());
+ metrics.getUnalignedCheckpoint(),
+ true);
statsCallback.reportSubtaskStats(vertex.getJobvertexId(), subtaskStateStats);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java
index 49a81f3..28fc832 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStats.java
@@ -187,8 +187,10 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
TaskStateStats taskStateStats = taskStats.get(jobVertexId);
if (taskStateStats != null && taskStateStats.reportSubtaskStats(subtask)) {
- currentNumAcknowledgedSubtasks++;
- latestAcknowledgedSubtask = subtask;
+ if (subtask.isCompleted()) {
+ currentNumAcknowledgedSubtasks++;
+ latestAcknowledgedSubtask = subtask;
+ }
currentStateSize += subtask.getStateSize();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskStateStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskStateStats.java
index 56eed10..bd7719b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskStateStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskStateStats.java
@@ -61,6 +61,9 @@ public class SubtaskStateStats implements Serializable {
/** Is the checkpoint completed as an unaligned checkpoint. */
private final boolean unalignedCheckpoint;
+ /** Is the checkpoint completed by this subtask. */
+ private final boolean completed;
+
SubtaskStateStats(
int subtaskIndex,
long ackTimestamp,
@@ -71,7 +74,8 @@ public class SubtaskStateStats implements Serializable {
long persistedData,
long alignmentDuration,
long checkpointStartDelay,
- boolean unalignedCheckpoint) {
+ boolean unalignedCheckpoint,
+ boolean completed) {
checkArgument(subtaskIndex >= 0, "Negative subtask index");
this.subtaskIndex = subtaskIndex;
@@ -85,6 +89,7 @@ public class SubtaskStateStats implements Serializable {
this.alignmentDuration = alignmentDuration;
this.checkpointStartDelay = checkpointStartDelay;
this.unalignedCheckpoint = unalignedCheckpoint;
+ this.completed = completed;
}
public int getSubtaskIndex() {
@@ -163,4 +168,8 @@ public class SubtaskStateStats implements Serializable {
public boolean getUnalignedCheckpoint() {
return unalignedCheckpoint;
}
+
+ public boolean isCompleted() {
+ return completed;
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateStats.java
index e151338..b4ed924 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateStats.java
@@ -64,8 +64,10 @@ public class TaskStateStats implements Serializable {
if (subtaskStats[subtaskIndex] == null) {
subtaskStats[subtaskIndex] = subtask;
- latestAckedSubtaskStats = subtask;
- numAcknowledgedSubtasks++;
+ if (subtask.isCompleted()) {
+ latestAckedSubtaskStats = subtask;
+ numAcknowledgedSubtasks++;
+ }
summaryStats.updateSummary(subtask);
@@ -175,7 +177,9 @@ public class TaskStateStats implements Serializable {
void updateSummary(SubtaskStateStats subtaskStats) {
stateSize.add(subtaskStats.getStateSize());
- ackTimestamp.add(subtaskStats.getAckTimestamp());
+ if (subtaskStats.isCompleted()) {
+ ackTimestamp.add(subtaskStats.getAckTimestamp());
+ }
syncCheckpointDuration.add(subtaskStats.getSyncCheckpointDuration());
asyncCheckpointDuration.add(subtaskStats.getAsyncCheckpointDuration());
processedData.add(subtaskStats.getProcessedData());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 2eb0318..b8418e8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -512,6 +512,17 @@ public class JobMaster extends PermanentlyFencedRpcEndpoint<JobMasterId>
jobID, executionAttemptID, checkpointId, checkpointMetrics, checkpointState);
}
+ @Override
+ public void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics) {
+
+ schedulerNG.reportCheckpointMetrics(
+ jobID, executionAttemptID, checkpointId, checkpointMetrics);
+ }
+
// TODO: This method needs a leader session ID
@Override
public void declineCheckpoint(DeclineCheckpoint decline) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java
index 716545d..d512c58 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java
@@ -210,7 +210,8 @@ public class TaskCheckpointStatisticDetailsHandler
subtask.getPersistedData(),
subtask.getAlignmentDuration()),
subtask.getCheckpointStartDelay(),
- subtask.getUnalignedCheckpoint()));
+ subtask.getUnalignedCheckpoint(),
+ !subtask.isCompleted()));
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/SubtaskCheckpointStatistics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/SubtaskCheckpointStatistics.java
index 4a20e9a..c8f9e87 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/SubtaskCheckpointStatistics.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/SubtaskCheckpointStatistics.java
@@ -108,6 +108,8 @@ public class SubtaskCheckpointStatistics {
public static final String FIELD_NAME_UNALIGNED_CHECKPOINT = "unaligned_checkpoint";
+ public static final String FIELD_NAME_ABORTED = "aborted";
+
@JsonProperty(FIELD_NAME_ACK_TIMESTAMP)
private final long ackTimestamp;
@@ -129,6 +131,9 @@ public class SubtaskCheckpointStatistics {
@JsonProperty(FIELD_NAME_UNALIGNED_CHECKPOINT)
private boolean unalignedCheckpoint;
+ @JsonProperty(FIELD_NAME_ABORTED)
+ private final boolean aborted;
+
@JsonCreator
public CompletedSubtaskCheckpointStatistics(
@JsonProperty(FIELD_NAME_INDEX) int index,
@@ -138,7 +143,8 @@ public class SubtaskCheckpointStatistics {
@JsonProperty(FIELD_NAME_CHECKPOINT_DURATION) CheckpointDuration checkpointDuration,
@JsonProperty(FIELD_NAME_ALIGNMENT) CheckpointAlignment alignment,
@JsonProperty(FIELD_NAME_START_DELAY) long startDelay,
- @JsonProperty(FIELD_NAME_UNALIGNED_CHECKPOINT) boolean unalignedCheckpoint) {
+ @JsonProperty(FIELD_NAME_UNALIGNED_CHECKPOINT) boolean unalignedCheckpoint,
+ @JsonProperty(FIELD_NAME_ABORTED) boolean aborted) {
super(index, "completed");
this.ackTimestamp = ackTimestamp;
this.duration = duration;
@@ -147,6 +153,7 @@ public class SubtaskCheckpointStatistics {
this.alignment = alignment;
this.startDelay = startDelay;
this.unalignedCheckpoint = unalignedCheckpoint;
+ this.aborted = aborted;
}
public long getAckTimestamp() {
@@ -177,6 +184,10 @@ public class SubtaskCheckpointStatistics {
return unalignedCheckpoint;
}
+ public boolean isAborted() {
+ return aborted;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -192,7 +203,8 @@ public class SubtaskCheckpointStatistics {
&& Objects.equals(checkpointDuration, that.checkpointDuration)
&& Objects.equals(alignment, that.alignment)
&& startDelay == that.startDelay
- && unalignedCheckpoint == that.unalignedCheckpoint;
+ && unalignedCheckpoint == that.unalignedCheckpoint
+ && aborted == that.aborted;
}
@Override
@@ -204,7 +216,8 @@ public class SubtaskCheckpointStatistics {
checkpointDuration,
alignment,
startDelay,
- unalignedCheckpoint);
+ unalignedCheckpoint,
+ aborted);
}
/** Duration of the checkpoint. */
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
index 3770643..45c8076 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
@@ -1016,6 +1016,34 @@ public abstract class SchedulerBase implements SchedulerNG {
}
@Override
+ public void reportCheckpointMetrics(
+ JobID jobID, ExecutionAttemptID attemptId, long id, CheckpointMetrics metrics) {
+ mainThreadExecutor.assertRunningInMainThread();
+
+ final CheckpointCoordinator checkpointCoordinator =
+ executionGraph.getCheckpointCoordinator();
+
+ if (checkpointCoordinator != null) {
+ ioExecutor.execute(
+ () -> {
+ try {
+ checkpointCoordinator.reportStats(id, attemptId, metrics);
+ } catch (Throwable t) {
+ log.warn("Error while processing report checkpoint stats message", t);
+ }
+ });
+ } else {
+ String errorMessage =
+ "Received ReportCheckpointStats message for job {} with no CheckpointCoordinator";
+ if (executionGraph.getState() == JobStatus.RUNNING) {
+ log.error(errorMessage, jobGraph.getJobID());
+ } else {
+ log.debug(errorMessage, jobGraph.getJobID());
+ }
+ }
+ }
+
+ @Override
public void declineCheckpoint(final DeclineCheckpoint decline) {
mainThreadExecutor.assertRunningInMainThread();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java
index 1cf57a7..5db4c5d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java
@@ -136,6 +136,12 @@ public interface SchedulerNG {
CheckpointMetrics checkpointMetrics,
TaskStateSnapshot checkpointState);
+ void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics);
+
void declineCheckpoint(DeclineCheckpoint decline);
CompletableFuture<String> stopWithSavepoint(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManager.java
index f1843be..2aed01e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManager.java
@@ -57,6 +57,15 @@ public interface TaskStateManager extends CheckpointListener, AutoCloseable {
@Nullable TaskStateSnapshot localState);
/**
+ * Report the stats for state snapshots for an aborted checkpoint.
+ *
+ * @param checkpointMetaData meta data from the checkpoint request.
+ * @param checkpointMetrics task level metrics for the checkpoint.
+ */
+ void reportIncompleteTaskStateSnapshots(
+ CheckpointMetaData checkpointMetaData, CheckpointMetrics checkpointMetrics);
+
+ /**
* Returns means to restore previously reported state of an operator running in the owning task.
*
* @param operatorID the id of the operator for which we request state.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManagerImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManagerImpl.java
index 2156afb..1474e14 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManagerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskStateManagerImpl.java
@@ -122,6 +122,14 @@ public class TaskStateManagerImpl implements TaskStateManager {
jobId, executionAttemptID, checkpointId, checkpointMetrics, acknowledgedState);
}
+ @Override
+ public void reportIncompleteTaskStateSnapshots(
+ CheckpointMetaData checkpointMetaData, CheckpointMetrics checkpointMetrics) {
+
+ checkpointResponder.reportCheckpointMetrics(
+ jobId, executionAttemptID, checkpointMetaData.getCheckpointId(), checkpointMetrics);
+ }
+
@Nonnull
@Override
public PrioritizedOperatorSubtaskState prioritizedOperatorState(OperatorID operatorID) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
index b445941..a712bba 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
@@ -49,6 +49,16 @@ public class RpcCheckpointResponder implements CheckpointResponder {
}
@Override
+ public void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics) {
+ checkpointCoordinatorGateway.reportCheckpointMetrics(
+ jobID, executionAttemptID, checkpointId, checkpointMetrics);
+ }
+
+ @Override
public void declineCheckpoint(
JobID jobID,
ExecutionAttemptID executionAttemptID,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/CheckpointResponder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/CheckpointResponder.java
index 40ef25c..acee2b6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/CheckpointResponder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/CheckpointResponder.java
@@ -18,12 +18,14 @@
package org.apache.flink.runtime.taskmanager;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
/** Responder for checkpoint acknowledge and decline messages in the {@link Task}. */
+@Internal
public interface CheckpointResponder {
/**
@@ -43,6 +45,20 @@ public interface CheckpointResponder {
TaskStateSnapshot subtaskState);
/**
+ * Report metrics for the given checkpoint. Can be used upon receiving abortion notification.
+ *
+ * @param jobID Job ID of the running job
+ * @param executionAttemptID Execution attempt ID of the running task
+ * @param checkpointId Meta data for this checkpoint
+ * @param checkpointMetrics Metrics of this checkpoint
+ */
+ void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics);
+
+ /**
* Declines the given checkpoint.
*
* @param jobID Job ID of the running job
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index f36412a..29c1a21 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -61,10 +61,10 @@ import org.apache.flink.runtime.state.memory.NonPersistentMetadataCheckpointStor
import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.function.TriFunctionWithException;
import org.apache.flink.shaded.guava18.com.google.common.collect.Iterables;
-import org.apache.flink.util.function.TriFunctionWithException;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
@@ -126,6 +126,18 @@ import static org.mockito.Mockito.when;
public class CheckpointCoordinatorTest extends TestLogger {
@Test
+ public void testAbortedCheckpointStatsUpdatedAfterFailure() throws Exception {
+ JobID jobID = new JobID();
+ testReportStatsAfterFailure(
+ jobID,
+ 1L,
+ (coordinator, attemptID, metrics) -> {
+ coordinator.reportStats(1L, attemptID, metrics);
+ return null;
+ });
+ }
+
+ @Test
public void testCheckpointStatsUpdatedAfterFailure() throws Exception {
JobID jobID = new JobID();
testReportStatsAfterFailure(
@@ -197,17 +209,28 @@ public class CheckpointCoordinatorTest extends TestLogger {
assertStatsEqual(
checkpointId,
+ lateReportVertex.getJobvertexId(),
+ 0,
lateReportedMetrics,
statsTracker.createSnapshot().getHistory().getCheckpointById(checkpointId));
}
private void assertStatsEqual(
- long checkpointId, CheckpointMetrics expected, AbstractCheckpointStats actual) {
+ long checkpointId,
+ JobVertexID jobVertexID,
+ int subtasIdx,
+ CheckpointMetrics expected,
+ AbstractCheckpointStats actual) {
assertEquals(checkpointId, actual.getCheckpointId());
assertEquals(CheckpointStatsStatus.FAILED, actual.getStatus());
- assertEquals(1, actual.getNumberOfAcknowledgedSubtasks());
assertEquals(expected.getTotalBytesPersisted(), actual.getStateSize());
- SubtaskStateStats taskStats = actual.getLatestAcknowledgedSubtaskStats();
+ assertEquals(0, actual.getNumberOfAcknowledgedSubtasks());
+ SubtaskStateStats taskStats =
+ actual.getAllTaskStateStats().stream()
+ .filter(s -> s.getJobVertexId().equals(jobVertexID))
+ .findAny()
+ .get()
+ .getSubtaskStats()[subtasIdx];
assertEquals(
expected.getAlignmentDurationNanos() / 1_000_000, taskStats.getAlignmentDuration());
assertEquals(expected.getUnalignedCheckpoint(), taskStats.getUnalignedCheckpoint());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java
index f227044..070f996 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java
@@ -445,7 +445,8 @@ public class CheckpointStatsTrackerTest {
persistedData,
ignored,
ignored,
- false);
+ false,
+ true);
assertTrue(pending.reportSubtaskStats(vertexID, subtaskStats));
@@ -518,6 +519,6 @@ public class CheckpointStatsTrackerTest {
// ------------------------------------------------------------------------
private SubtaskStateStats createSubtaskStats(int index) {
- return new SubtaskStateStats(index, 0, 0, 0, 0, 0, 0, 0, 0, false);
+ return new SubtaskStateStats(index, 0, 0, 0, 0, 0, 0, 0, 0, false, true);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java
index 22caf2f..d1bb382 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointTest.java
@@ -379,7 +379,7 @@ public class CompletedCheckpointTest {
123129837912L,
42L,
44L,
- new SubtaskStateStats(123, 213123, 123123, 0, 0, 0, 0, 0, 0, false),
+ new SubtaskStateStats(123, 213123, 123123, 0, 0, 0, 0, 0, 0, false, true),
null);
CompletedCheckpointStats copy = CommonTestUtils.createCopySerializable(completed);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStatsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStatsTest.java
index b0ee97f..80f4aec 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStatsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointStatsTest.java
@@ -295,6 +295,7 @@ public class PendingCheckpointStatsTest {
Integer.MAX_VALUE + (long) index,
Integer.MAX_VALUE + (long) index,
Integer.MAX_VALUE + (long) index,
- false);
+ false,
+ true);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SubtaskStateStatsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SubtaskStateStatsTest.java
index 770520f..26092e0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SubtaskStateStatsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SubtaskStateStatsTest.java
@@ -50,7 +50,8 @@ public class SubtaskStateStatsTest {
Integer.MAX_VALUE + 9L,
Integer.MAX_VALUE + 6L,
Integer.MAX_VALUE + 7L,
- false);
+ false,
+ true);
stats = serialize ? CommonTestUtils.createCopySerializable(stats) : stats;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TaskStateStatsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TaskStateStatsTest.java
index e53c19e..413a7f5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TaskStateStatsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/TaskStateStatsTest.java
@@ -76,7 +76,8 @@ public class TaskStateStatsTest {
rand.nextInt(128),
rand.nextInt(128),
rand.nextInt(128),
- false);
+ false,
+ true);
stateSize += subtasks[i].getStateSize();
processedData += subtasks[i].getProcessedData();
@@ -97,7 +98,7 @@ public class TaskStateStatsTest {
assertFalse(
taskStats.reportSubtaskStats(
- new SubtaskStateStats(0, 0, 0, 0, 0, 0, 0, 0, 0, false)));
+ new SubtaskStateStats(0, 0, 0, 0, 0, 0, 0, 0, 0, false, true)));
taskStats = serialize ? CommonTestUtils.createCopySerializable(taskStats) : taskStats;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
index e5a43e1..a6b4155 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
@@ -435,6 +435,13 @@ public class TestingJobMasterGateway implements JobMasterGateway {
}
@Override
+ public void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics) {}
+
+ @Override
public JobMasterId getFencingToken() {
return fencingTokenSupplier.get();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/TaskCheckpointStatisticsWithSubtaskDetailsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/TaskCheckpointStatisticsWithSubtaskDetailsTest.java
index 6aca236..1270040 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/TaskCheckpointStatisticsWithSubtaskDetailsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/TaskCheckpointStatisticsWithSubtaskDetailsTest.java
@@ -65,7 +65,8 @@ public class TaskCheckpointStatisticsWithSubtaskDetailsTest
new SubtaskCheckpointStatistics.CompletedSubtaskCheckpointStatistics
.CheckpointAlignment(2L, 4L, 5L, 3L),
42L,
- true));
+ true,
+ false));
return new TaskCheckpointStatisticsWithSubtaskDetails(
4L,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java
index 1bf5cc4e..71a990f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java
@@ -208,6 +208,13 @@ public class TestingSchedulerNG implements SchedulerNG {
return null;
}
+ @Override
+ public void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics) {}
+
public static Builder newBuilder() {
return new Builder();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TestTaskStateManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TestTaskStateManager.java
index 8ac850a..cb3f64d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TestTaskStateManager.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TestTaskStateManager.java
@@ -117,6 +117,10 @@ public class TestTaskStateManager implements TaskStateManager {
}
}
+ @Override
+ public void reportIncompleteTaskStateSnapshots(
+ CheckpointMetaData checkpointMetaData, CheckpointMetrics checkpointMetrics) {}
+
@Nonnull
@Override
public PrioritizedOperatorSubtaskState prioritizedOperatorState(OperatorID operatorID) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpCheckpointResponder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpCheckpointResponder.java
index 406792e..f0f9b70 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpCheckpointResponder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/NoOpCheckpointResponder.java
@@ -33,4 +33,11 @@ public enum NoOpCheckpointResponder implements CheckpointResponder {
@Override
public void declineCheckpoint(JobID j, ExecutionAttemptID e, long l, Throwable t) {}
+
+ @Override
+ public void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics) {}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestCheckpointResponder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestCheckpointResponder.java
index 5e1d472..057e6e3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestCheckpointResponder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestCheckpointResponder.java
@@ -61,6 +61,13 @@ public class TestCheckpointResponder implements CheckpointResponder {
}
@Override
+ public void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics) {}
+
+ @Override
public void declineCheckpoint(
JobID jobID,
ExecutionAttemptID executionAttemptID,
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncSnapshotTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncSnapshotTest.java
index 9f1aec7..a7847a4 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncSnapshotTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncSnapshotTest.java
@@ -186,6 +186,13 @@ public class RocksDBAsyncSnapshotTest extends TestLogger {
}
@Override
+ public void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics) {}
+
+ @Override
public void declineCheckpoint(
JobID jobID,
ExecutionAttemptID executionAttemptID,
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
index 412af3e..cf31d62 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
@@ -21,6 +21,7 @@ import org.apache.flink.core.fs.FileSystemSafetyNet;
import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.execution.Environment;
@@ -214,6 +215,19 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
acknowledgedTaskStateSnapshot);
}
+ private void reportAbortedSnapshotStats(long stateSize) {
+ CheckpointMetrics metrics = checkpointMetrics.setTotalBytesPersisted(stateSize).build();
+ LOG.trace(
+ "{} - report failed checkpoint stats: {} {}",
+ taskName,
+ checkpointMetaData.getCheckpointId(),
+ metrics);
+
+ taskEnvironment
+ .getTaskStateManager()
+ .reportIncompleteTaskStateSnapshots(checkpointMetaData, metrics);
+ }
+
private void handleExecutionException(Exception e) {
boolean didCleanup = false;
@@ -275,7 +289,7 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
try {
final long stateSize = cleanup();
- LOG.debug("discarded state size: {}", stateSize);
+ reportAbortedSnapshotStats(stateSize);
} catch (Exception cleanupException) {
LOG.warn(
"Could not properly clean up the async checkpoint runnable.",
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
index 5fd78ba..7393129 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
@@ -264,6 +264,13 @@ public class TaskCheckpointingBehaviourTest extends TestLogger {
}
@Override
+ public void reportCheckpointMetrics(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics) {}
+
+ @Override
public void declineCheckpoint(
JobID jobID,
ExecutionAttemptID executionAttemptID,