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 2024/11/27 10:48:00 UTC
(flink) branch master updated: [FLINK-36764] Add checkpoint type and unaligned flag to the checkpoint trace
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
The following commit(s) were added to refs/heads/master by this push:
new 09ad45129ae [FLINK-36764] Add checkpoint type and unaligned flag to the checkpoint trace
09ad45129ae is described below
commit 09ad45129ae729f9c40bddfc50b34f3f3c04d48e
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Thu Nov 21 11:01:33 2024 +0200
[FLINK-36764] Add checkpoint type and unaligned flag to the checkpoint trace
---
docs/content.zh/docs/ops/traces.md | 12 +++++--
docs/content/docs/ops/traces.md | 12 +++++--
.../checkpoint/DefaultCheckpointStatsTracker.java | 8 ++++-
.../DefaultCheckpointStatsTrackerTest.java | 39 +++++++++++++++++-----
4 files changed, 58 insertions(+), 13 deletions(-)
diff --git a/docs/content.zh/docs/ops/traces.md b/docs/content.zh/docs/ops/traces.md
index 8157d1ecddd..393551074f6 100644
--- a/docs/content.zh/docs/ops/traces.md
+++ b/docs/content.zh/docs/ops/traces.md
@@ -97,8 +97,8 @@ Flink reports a single span trace for the whole checkpoint and job initializatio
</thead>
<tbody>
<tr>
- <th rowspan="18">org.apache.flink.</br>runtime.checkpoint.</br>CheckpointStatsTracker</th>
- <th rowspan="6"><strong>Checkpoint</strong></th>
+ <th rowspan="20">org.apache.flink.</br>runtime.checkpoint.</br>CheckpointStatsTracker</th>
+ <th rowspan="8"><strong>Checkpoint</strong></th>
<td>startTs</td>
<td>Timestamp when the checkpoint has started.</td>
</tr>
@@ -122,6 +122,14 @@ Flink reports a single span trace for the whole checkpoint and job initializatio
<td>checkpointStatus</td>
<td>What was the state of this checkpoint: FAILED or COMPLETED.</td>
</tr>
+ <tr>
+ <td>checkpointType</td>
+ <td>Type of the checkpoint. For example: "Checkpoint", "Full Checkpoint" or "Terminate Savepoint" ...</td>
+ </tr>
+ <tr>
+ <td>isUnaligned</td>
+ <td>Whether checkpoint was aligned or unaligned.</td>
+ </tr>
<tr>
<th rowspan="12"><strong>JobInitialization</strong></th>
<td>startTs</td>
diff --git a/docs/content/docs/ops/traces.md b/docs/content/docs/ops/traces.md
index 8157d1ecddd..393551074f6 100644
--- a/docs/content/docs/ops/traces.md
+++ b/docs/content/docs/ops/traces.md
@@ -97,8 +97,8 @@ Flink reports a single span trace for the whole checkpoint and job initializatio
</thead>
<tbody>
<tr>
- <th rowspan="18">org.apache.flink.</br>runtime.checkpoint.</br>CheckpointStatsTracker</th>
- <th rowspan="6"><strong>Checkpoint</strong></th>
+ <th rowspan="20">org.apache.flink.</br>runtime.checkpoint.</br>CheckpointStatsTracker</th>
+ <th rowspan="8"><strong>Checkpoint</strong></th>
<td>startTs</td>
<td>Timestamp when the checkpoint has started.</td>
</tr>
@@ -122,6 +122,14 @@ Flink reports a single span trace for the whole checkpoint and job initializatio
<td>checkpointStatus</td>
<td>What was the state of this checkpoint: FAILED or COMPLETED.</td>
</tr>
+ <tr>
+ <td>checkpointType</td>
+ <td>Type of the checkpoint. For example: "Checkpoint", "Full Checkpoint" or "Terminate Savepoint" ...</td>
+ </tr>
+ <tr>
+ <td>isUnaligned</td>
+ <td>Whether checkpoint was aligned or unaligned.</td>
+ </tr>
<tr>
<th rowspan="12"><strong>JobInitialization</strong></th>
<td>startTs</td>
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTracker.java
index b74a8d8d307..fa42bb9742f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTracker.java
@@ -256,7 +256,13 @@ public class DefaultCheckpointStatsTracker implements CheckpointStatsTracker {
.setAttribute("checkpointId", checkpointStats.getCheckpointId())
.setAttribute("fullSize", checkpointStats.getStateSize())
.setAttribute("checkpointedSize", checkpointStats.getCheckpointedSize())
- .setAttribute("checkpointStatus", checkpointStats.getStatus().name()));
+ .setAttribute("checkpointStatus", checkpointStats.getStatus().name())
+ .setAttribute(
+ "isUnaligned",
+ Boolean.toString(checkpointStats.isUnalignedCheckpoint()))
+ .setAttribute(
+ "checkpointType",
+ checkpointStats.getProperties().getCheckpointType().getName()));
if (LOG.isDebugEnabled()) {
StringWriter sw = new StringWriter();
MAPPER.writeValue(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTrackerTest.java
index 86c6f8f6e8b..1c2d7d0e26e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTrackerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointStatsTrackerTest.java
@@ -48,7 +48,6 @@ import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
-import java.util.stream.Collectors;
import static java.util.Collections.singletonMap;
import static org.assertj.core.api.Assertions.assertThat;
@@ -379,17 +378,37 @@ class DefaultCheckpointStatsTrackerTest {
CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
singletonMap(jobVertexID, 1));
- pending.reportSubtaskStats(jobVertexID, createSubtaskStats(0));
+ pending.reportSubtaskStats(jobVertexID, createSubtaskStats(0, false));
+
+ // Complete checkpoint => new snapshot
+ tracker.reportCompletedCheckpoint(pending.toCompletedCheckpointStats(null));
+
+ assertThat(reportedSpans.size()).isEqualTo(1);
+ Span reportedSpan = Iterables.getOnlyElement(reportedSpans);
+ assertThat(reportedSpan.getAttributes().get("checkpointId")).isEqualTo(42L);
+ assertThat(reportedSpan.getAttributes().get("checkpointType")).isEqualTo("Checkpoint");
+ assertThat(reportedSpan.getAttributes().get("isUnaligned")).isEqualTo("false");
+
+ reportedSpans.clear();
+
+ pending =
+ tracker.reportPendingCheckpoint(
+ 43,
+ 1,
+ CheckpointProperties.forCheckpoint(
+ CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
+ singletonMap(jobVertexID, 1));
+
+ pending.reportSubtaskStats(jobVertexID, createSubtaskStats(0, true));
// Complete checkpoint => new snapshot
tracker.reportCompletedCheckpoint(pending.toCompletedCheckpointStats(null));
assertThat(reportedSpans.size()).isEqualTo(1);
- assertThat(
- reportedSpans.stream()
- .map(span -> span.getAttributes().get("checkpointId"))
- .collect(Collectors.toList()))
- .containsExactly(42L);
+ reportedSpan = Iterables.getOnlyElement(reportedSpans);
+ assertThat(reportedSpan.getAttributes().get("checkpointId")).isEqualTo(43L);
+ assertThat(reportedSpan.getAttributes().get("checkpointType")).isEqualTo("Checkpoint");
+ assertThat(reportedSpan.getAttributes().get("isUnaligned")).isEqualTo("true");
}
@Test
@@ -763,7 +782,11 @@ class DefaultCheckpointStatsTrackerTest {
// ------------------------------------------------------------------------
private SubtaskStateStats createSubtaskStats(int index) {
- return new SubtaskStateStats(index, 0, 0, 0, 0, 0, 0, 0, 0, 0, false, true);
+ return createSubtaskStats(index, false);
+ }
+
+ private SubtaskStateStats createSubtaskStats(int index, boolean unaligned) {
+ return new SubtaskStateStats(index, 0, 0, 0, 0, 0, 0, 0, 0, 0, unaligned, true);
}
private void reportRestoredCheckpoint(