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 2023/05/10 07:41:59 UTC
[flink] branch master updated (400530d30af -> 35254cbdc9a)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 400530d30af [FLINK-31546][jdbc-driver] Close all statements when connection is closed
new 345ddf859f5 [hotfix] Migrate checkpoint statistics related tests to JUnit5 and AssertJ
new 35254cbdc9a [FLINK-31959][checkpoint] Correct the unaligned checkpoint type at checkpoint level
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../src/test/resources/rest_api_v1.snapshot | 8 +-
.../detail/job-checkpoints-detail.component.ts | 8 +-
.../checkpoint/AbstractCheckpointStats.java | 3 +
.../checkpoint/CompletedCheckpointStats.java | 12 ++
.../runtime/checkpoint/FailedCheckpointStats.java | 3 +
.../runtime/checkpoint/PendingCheckpointStats.java | 15 ++
.../messages/checkpoints/CheckpointStatistics.java | 22 ++-
.../CompletedCheckpointStatsSummaryTest.java | 65 ++++----
.../checkpoint/CompletedCheckpointTest.java | 81 +++++----
.../checkpoint/FailedCheckpointStatsTest.java | 42 ++---
.../checkpoint/PendingCheckpointStatsTest.java | 185 +++++++++++----------
.../checkpoints/CheckpointingStatisticsTest.java | 50 +++++-
12 files changed, 291 insertions(+), 203 deletions(-)
[flink] 01/02: [hotfix] Migrate checkpoint statistics related tests to JUnit5 and AssertJ
Posted by pn...@apache.org.
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 345ddf859f59f60ff07fc62fff031be630d22092
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Fri Apr 28 11:25:06 2023 +0800
[hotfix] Migrate checkpoint statistics related tests to JUnit5 and AssertJ
---
.../CompletedCheckpointStatsSummaryTest.java | 55 +++----
.../checkpoint/CompletedCheckpointTest.java | 78 +++++-----
.../checkpoint/FailedCheckpointStatsTest.java | 39 +++--
.../checkpoint/PendingCheckpointStatsTest.java | 160 ++++++++++-----------
4 files changed, 159 insertions(+), 173 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStatsSummaryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStatsSummaryTest.java
index da8a37e30ed..4ae0e14163b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStatsSummaryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStatsSummaryTest.java
@@ -21,13 +21,14 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.Map;
import static java.util.Collections.singletonMap;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.data.Offset.offset;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -35,7 +36,7 @@ public class CompletedCheckpointStatsSummaryTest {
/** Tests simple updates of the completed checkpoint stats. */
@Test
- public void testSimpleUpdates() throws Exception {
+ void testSimpleUpdates() {
long triggerTimestamp = 123123L;
long ackTimestamp = 123123 + 1212312399L;
long stateSize = Integer.MAX_VALUE + 17787L;
@@ -43,10 +44,10 @@ public class CompletedCheckpointStatsSummaryTest {
long persistedData = Integer.MAX_VALUE + 42L;
CompletedCheckpointStatsSummary summary = new CompletedCheckpointStatsSummary();
- assertEquals(0, summary.getStateSizeStats().getCount());
- assertEquals(0, summary.getEndToEndDurationStats().getCount());
- assertEquals(0, summary.getProcessedDataStats().getCount());
- assertEquals(0, summary.getPersistedDataStats().getCount());
+ assertThat(summary.getStateSizeStats().getCount()).isZero();
+ assertThat(summary.getEndToEndDurationStats().getCount()).isZero();
+ assertThat(summary.getProcessedDataStats().getCount()).isZero();
+ assertThat(summary.getPersistedDataStats().getCount()).isZero();
int numCheckpoints = 10;
@@ -62,28 +63,28 @@ public class CompletedCheckpointStatsSummaryTest {
summary.updateSummary(completed);
- assertEquals(i + 1, summary.getStateSizeStats().getCount());
- assertEquals(i + 1, summary.getEndToEndDurationStats().getCount());
- assertEquals(i + 1, summary.getProcessedDataStats().getCount());
- assertEquals(i + 1, summary.getPersistedDataStats().getCount());
+ assertThat(summary.getStateSizeStats().getCount()).isEqualTo(i + 1);
+ assertThat(summary.getEndToEndDurationStats().getCount()).isEqualTo(i + 1);
+ assertThat(summary.getProcessedDataStats().getCount()).isEqualTo(i + 1);
+ assertThat(summary.getPersistedDataStats().getCount()).isEqualTo(i + 1);
}
StatsSummary stateSizeStats = summary.getStateSizeStats();
- assertEquals(stateSize, stateSizeStats.getMinimum());
- assertEquals(stateSize + numCheckpoints - 1, stateSizeStats.getMaximum());
+ assertThat(stateSizeStats.getMinimum()).isEqualTo(stateSize);
+ assertThat(stateSizeStats.getMaximum()).isEqualTo(stateSize + numCheckpoints - 1);
StatsSummary durationStats = summary.getEndToEndDurationStats();
- assertEquals(ackTimestamp - triggerTimestamp, durationStats.getMinimum());
- assertEquals(
- ackTimestamp - triggerTimestamp + numCheckpoints - 1, durationStats.getMaximum());
+ assertThat(durationStats.getMinimum()).isEqualTo(ackTimestamp - triggerTimestamp);
+ assertThat(durationStats.getMaximum())
+ .isEqualTo(ackTimestamp - triggerTimestamp + numCheckpoints - 1);
StatsSummary processedDataStats = summary.getProcessedDataStats();
- assertEquals(processedData, processedDataStats.getMinimum());
- assertEquals(processedData + numCheckpoints - 1, processedDataStats.getMaximum());
+ assertThat(processedDataStats.getMinimum()).isEqualTo(processedData);
+ assertThat(processedDataStats.getMaximum()).isEqualTo(processedData + numCheckpoints - 1);
StatsSummary persistedDataStats = summary.getPersistedDataStats();
- assertEquals(persistedData, persistedDataStats.getMinimum());
- assertEquals(persistedData + numCheckpoints - 1, persistedDataStats.getMaximum());
+ assertThat(persistedDataStats.getMinimum()).isEqualTo(persistedData);
+ assertThat(persistedDataStats.getMaximum()).isEqualTo(persistedData + numCheckpoints - 1);
}
private CompletedCheckpointStats createCompletedCheckpoint(
@@ -118,7 +119,7 @@ public class CompletedCheckpointStatsSummaryTest {
/** Simply test that quantiles can be computed and fields are not permuted. */
@Test
- public void testQuantiles() {
+ void testQuantiles() {
int stateSize = 100;
int processedData = 200;
int persistedData = 300;
@@ -140,10 +141,12 @@ public class CompletedCheckpointStatsSummaryTest {
new SubtaskStateStats(0, lastAck),
""));
CompletedCheckpointStatsSummarySnapshot snapshot = summary.createSnapshot();
- assertEquals(stateSize, snapshot.getStateSizeStats().getQuantile(1), 0);
- assertEquals(processedData, snapshot.getProcessedDataStats().getQuantile(1), 0);
- assertEquals(persistedData, snapshot.getPersistedDataStats().getQuantile(1), 0);
- assertEquals(
- lastAck - triggerTimestamp, snapshot.getEndToEndDurationStats().getQuantile(1), 0);
+ assertThat(snapshot.getStateSizeStats().getQuantile(1)).isCloseTo(stateSize, offset(0d));
+ assertThat(snapshot.getProcessedDataStats().getQuantile(1))
+ .isCloseTo(processedData, offset(0d));
+ assertThat(snapshot.getPersistedDataStats().getQuantile(1))
+ .isCloseTo(persistedData, offset(0d));
+ assertThat(snapshot.getEndToEndDurationStats().getQuantile(1))
+ .isCloseTo(lastAck - triggerTimestamp, offset(0d));
}
}
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 a0b67cd0dfa..2f6b7259cf2 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
@@ -29,9 +29,7 @@ import org.apache.flink.runtime.state.SharedStateRegistryImpl;
import org.apache.flink.runtime.state.testutils.EmptyStreamStateHandle;
import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Collections;
@@ -39,9 +37,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -49,10 +45,8 @@ import static org.mockito.Mockito.verify;
/** Unit tests for the {@link CompletedCheckpoint}. */
public class CompletedCheckpointTest {
- @Rule public final TemporaryFolder tmpFolder = new TemporaryFolder();
-
@Test
- public void testCompareCheckpointsWithDifferentOrder() {
+ void testCompareCheckpointsWithDifferentOrder() {
CompletedCheckpoint checkpoint1 =
new CompletedCheckpoint(
@@ -90,11 +84,11 @@ public class CompletedCheckpointTest {
checkpoints2.add(checkpoint1);
checkpoints2.add(checkpoint2);
- assertFalse(CompletedCheckpoint.checkpointsMatch(checkpoints1, checkpoints2));
+ assertThat(CompletedCheckpoint.checkpointsMatch(checkpoints1, checkpoints2)).isFalse();
}
@Test
- public void testCompareCheckpointsWithSameOrder() {
+ void testCompareCheckpointsWithSameOrder() {
CompletedCheckpoint checkpoint1 =
new CompletedCheckpoint(
@@ -132,12 +126,12 @@ public class CompletedCheckpointTest {
checkpoints2.add(checkpoint2);
checkpoints2.add(checkpoint1);
- assertTrue(CompletedCheckpoint.checkpointsMatch(checkpoints1, checkpoints2));
+ assertThat(CompletedCheckpoint.checkpointsMatch(checkpoints1, checkpoints2)).isTrue();
}
/** Verify that both JobID and checkpoint id are taken into account when comparing. */
@Test
- public void testCompareCheckpointsWithSameJobID() {
+ void testCompareCheckpointsWithSameJobID() {
JobID jobID = new JobID();
CompletedCheckpoint checkpoint1 =
@@ -172,12 +166,12 @@ public class CompletedCheckpointTest {
List<CompletedCheckpoint> checkpoints2 = new ArrayList<>();
checkpoints2.add(checkpoint2);
- assertFalse(CompletedCheckpoint.checkpointsMatch(checkpoints1, checkpoints2));
+ assertThat(CompletedCheckpoint.checkpointsMatch(checkpoints1, checkpoints2)).isFalse();
}
/** Verify that both JobID and checkpoint id are taken into account when comparing. */
@Test
- public void testCompareCheckpointsWithSameCheckpointId() {
+ void testCompareCheckpointsWithSameCheckpointId() {
JobID jobID1 = new JobID();
JobID jobID2 = new JobID();
@@ -213,11 +207,11 @@ public class CompletedCheckpointTest {
List<CompletedCheckpoint> checkpoints2 = new ArrayList<>();
checkpoints2.add(checkpoint2);
- assertFalse(CompletedCheckpoint.checkpointsMatch(checkpoints1, checkpoints2));
+ assertThat(CompletedCheckpoint.checkpointsMatch(checkpoints1, checkpoints2)).isFalse();
}
@Test
- public void testRegisterStatesAtRegistry() {
+ void testRegisterStatesAtRegistry() {
OperatorState state = mock(OperatorState.class);
Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
operatorStates.put(new OperatorID(), state);
@@ -242,7 +236,7 @@ public class CompletedCheckpointTest {
/** Tests that the garbage collection properties are respected when subsuming checkpoints. */
@Test
- public void testCleanUpOnSubsume() throws Exception {
+ void testCleanUpOnSubsume() throws Exception {
OperatorState state = mock(OperatorState.class);
Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
operatorStates.put(new OperatorID(), state);
@@ -275,13 +269,13 @@ public class CompletedCheckpointTest {
checkpoint.markAsDiscardedOnSubsume().discard();
verify(state, times(1)).discardState();
- assertTrue(location.isDisposed());
- assertTrue(metadata.isDisposed());
+ assertThat(location.isDisposed()).isTrue();
+ assertThat(metadata.isDisposed()).isTrue();
}
/** Tests that the garbage collection properties are respected when shutting down. */
@Test
- public void testCleanUpOnShutdown() throws Exception {
+ void testCleanUpOnShutdown() throws Exception {
JobStatus[] terminalStates =
new JobStatus[] {
JobStatus.FINISHED, JobStatus.CANCELED, JobStatus.FAILED, JobStatus.SUSPENDED
@@ -323,8 +317,8 @@ public class CompletedCheckpointTest {
checkpoint.markAsDiscardedOnShutdown(status).discard();
verify(state, times(0)).discardState();
- assertFalse(retainedLocation.isDisposed());
- assertFalse(retainedHandle.isDisposed());
+ assertThat(retainedLocation.isDisposed()).isFalse();
+ assertThat(retainedHandle.isDisposed()).isFalse();
// Discard
EmptyStreamStateHandle discardHandle = new EmptyStreamStateHandle();
@@ -351,14 +345,14 @@ public class CompletedCheckpointTest {
checkpoint.markAsDiscardedOnShutdown(status).discard();
verify(state, times(1)).discardState();
- assertTrue(discardLocation.isDisposed());
- assertTrue(discardHandle.isDisposed());
+ assertThat(discardLocation.isDisposed()).isTrue();
+ assertThat(discardHandle.isDisposed()).isTrue();
}
}
/** Tests that the stats callbacks happen if the callback is registered. */
@Test
- public void testCompletedCheckpointStatsCallbacks() throws Exception {
+ void testCompletedCheckpointStatsCallbacks() throws Exception {
Map<JobVertexID, TaskStateStats> taskStats = new HashMap<>();
JobVertexID jobVertexId = new JobVertexID();
taskStats.put(jobVertexId, new TaskStateStats(jobVertexId, 1));
@@ -390,11 +384,11 @@ public class CompletedCheckpointTest {
checkpointStats);
completed.markAsDiscardedOnShutdown(JobStatus.FINISHED).discard();
- assertTrue(checkpointStats.isDiscarded());
+ assertThat(checkpointStats.isDiscarded()).isTrue();
}
@Test
- public void testIsJavaSerializable() throws Exception {
+ void testIsJavaSerializable() throws Exception {
TaskStateStats task1 = new TaskStateStats(new JobVertexID(), 3);
TaskStateStats task2 = new TaskStateStats(new JobVertexID(), 4);
@@ -420,20 +414,18 @@ public class CompletedCheckpointTest {
CompletedCheckpointStats copy = CommonTestUtils.createCopySerializable(completed);
- assertEquals(completed.getCheckpointId(), copy.getCheckpointId());
- assertEquals(completed.getTriggerTimestamp(), copy.getTriggerTimestamp());
- assertEquals(completed.getProperties(), copy.getProperties());
- assertEquals(completed.getNumberOfSubtasks(), copy.getNumberOfSubtasks());
- assertEquals(
- completed.getNumberOfAcknowledgedSubtasks(),
- copy.getNumberOfAcknowledgedSubtasks());
- assertEquals(completed.getEndToEndDuration(), copy.getEndToEndDuration());
- assertEquals(completed.getStateSize(), copy.getStateSize());
- assertEquals(completed.getProcessedData(), copy.getProcessedData());
- assertEquals(completed.getPersistedData(), copy.getPersistedData());
- assertEquals(
- completed.getLatestAcknowledgedSubtaskStats().getSubtaskIndex(),
- copy.getLatestAcknowledgedSubtaskStats().getSubtaskIndex());
- assertEquals(completed.getStatus(), copy.getStatus());
+ assertThat(copy.getCheckpointId()).isEqualTo(completed.getCheckpointId());
+ assertThat(copy.getTriggerTimestamp()).isEqualTo(completed.getTriggerTimestamp());
+ assertThat(copy.getProperties()).isEqualTo(completed.getProperties());
+ assertThat(copy.getNumberOfSubtasks()).isEqualTo(completed.getNumberOfSubtasks());
+ assertThat(copy.getNumberOfAcknowledgedSubtasks())
+ .isEqualTo(completed.getNumberOfAcknowledgedSubtasks());
+ assertThat(copy.getEndToEndDuration()).isEqualTo(completed.getEndToEndDuration());
+ assertThat(copy.getStateSize()).isEqualTo(completed.getStateSize());
+ assertThat(copy.getProcessedData()).isEqualTo(completed.getProcessedData());
+ assertThat(copy.getPersistedData()).isEqualTo(completed.getPersistedData());
+ assertThat(copy.getLatestAcknowledgedSubtaskStats().getSubtaskIndex())
+ .isEqualTo(completed.getLatestAcknowledgedSubtaskStats().getSubtaskIndex());
+ assertThat(copy.getStatus()).isEqualTo(completed.getStatus());
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java
index 069ab2dd58d..28beabc3e05 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java
@@ -21,13 +21,13 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.io.NotSerializableException;
import java.util.HashMap;
import java.util.Map;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
public class FailedCheckpointStatsTest {
@@ -35,7 +35,7 @@ public class FailedCheckpointStatsTest {
* Tests that the end to end duration of a failed checkpoint is the duration until the failure.
*/
@Test
- public void testEndToEndDuration() throws Exception {
+ void testEndToEndDuration() {
long duration = 123912931293L;
long triggerTimestamp = 10123;
long failureTimestamp = triggerTimestamp + duration;
@@ -61,11 +61,11 @@ public class FailedCheckpointStatsTest {
null,
null);
- assertEquals(duration, failed.getEndToEndDuration());
+ assertThat(failed.getEndToEndDuration()).isEqualTo(duration);
}
@Test
- public void testIsJavaSerializable() throws Exception {
+ void testIsJavaSerializable() throws Exception {
long duration = 123912931293L;
long triggerTimestamp = 10123;
long failureTimestamp = triggerTimestamp + duration;
@@ -93,20 +93,19 @@ public class FailedCheckpointStatsTest {
FailedCheckpointStats copy = CommonTestUtils.createCopySerializable(failed);
- assertEquals(failed.getCheckpointId(), copy.getCheckpointId());
- assertEquals(failed.getTriggerTimestamp(), copy.getTriggerTimestamp());
- assertEquals(failed.getProperties(), copy.getProperties());
- assertEquals(failed.getNumberOfSubtasks(), copy.getNumberOfSubtasks());
- assertEquals(
- failed.getNumberOfAcknowledgedSubtasks(), copy.getNumberOfAcknowledgedSubtasks());
- assertEquals(failed.getEndToEndDuration(), copy.getEndToEndDuration());
- assertEquals(failed.getStateSize(), copy.getStateSize());
- assertEquals(failed.getProcessedData(), copy.getProcessedData());
- assertEquals(failed.getPersistedData(), copy.getPersistedData());
- assertEquals(
- failed.getLatestAcknowledgedSubtaskStats(),
- copy.getLatestAcknowledgedSubtaskStats());
- assertEquals(failed.getStatus(), copy.getStatus());
- assertEquals(failed.getFailureMessage(), copy.getFailureMessage());
+ assertThat(copy.getCheckpointId()).isEqualTo(failed.getCheckpointId());
+ assertThat(copy.getTriggerTimestamp()).isEqualTo(failed.getTriggerTimestamp());
+ assertThat(copy.getProperties()).isEqualTo(failed.getProperties());
+ assertThat(copy.getNumberOfSubtasks()).isEqualTo(failed.getNumberOfSubtasks());
+ assertThat(copy.getNumberOfAcknowledgedSubtasks())
+ .isEqualTo(failed.getNumberOfAcknowledgedSubtasks());
+ assertThat(copy.getEndToEndDuration()).isEqualTo(failed.getEndToEndDuration());
+ assertThat(copy.getStateSize()).isEqualTo(failed.getStateSize());
+ assertThat(copy.getProcessedData()).isEqualTo(failed.getProcessedData());
+ assertThat(copy.getPersistedData()).isEqualTo(failed.getPersistedData());
+ assertThat(copy.getLatestAcknowledgedSubtaskStats())
+ .isEqualTo(failed.getLatestAcknowledgedSubtaskStats());
+ assertThat(copy.getStatus()).isEqualTo(failed.getStatus());
+ assertThat(copy.getFailureMessage()).isEqualTo(failed.getFailureMessage());
}
}
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 e46f6083b27..33d3313c767 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
@@ -21,16 +21,12 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import java.util.HashMap;
-import static junit.framework.TestCase.assertFalse;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -38,7 +34,7 @@ public class PendingCheckpointStatsTest {
/** Tests reporting of subtask stats. */
@Test
- public void testReportSubtaskStats() throws Exception {
+ void testReportSubtaskStats() {
long checkpointId = Integer.MAX_VALUE + 1222L;
long triggerTimestamp = Integer.MAX_VALUE - 1239L;
CheckpointProperties props =
@@ -57,22 +53,22 @@ public class PendingCheckpointStatsTest {
checkpointId, triggerTimestamp, props, totalSubtaskCount, taskStats);
// Check initial state
- assertEquals(checkpointId, pending.getCheckpointId());
- assertEquals(triggerTimestamp, pending.getTriggerTimestamp());
- assertEquals(props, pending.getProperties());
- assertEquals(CheckpointStatsStatus.IN_PROGRESS, pending.getStatus());
- assertEquals(0, pending.getNumberOfAcknowledgedSubtasks());
- assertEquals(0, pending.getStateSize());
- assertEquals(totalSubtaskCount, pending.getNumberOfSubtasks());
- assertNull(pending.getLatestAcknowledgedSubtaskStats());
- assertEquals(-1, pending.getLatestAckTimestamp());
- assertEquals(-1, pending.getEndToEndDuration());
- assertEquals(task1, pending.getTaskStateStats(task1.getJobVertexId()));
- assertEquals(task2, pending.getTaskStateStats(task2.getJobVertexId()));
- assertNull(pending.getTaskStateStats(new JobVertexID()));
+ assertThat(pending.getCheckpointId()).isEqualTo(checkpointId);
+ assertThat(pending.getTriggerTimestamp()).isEqualTo(triggerTimestamp);
+ assertThat(pending.getProperties()).isEqualTo(props);
+ assertThat(pending.getStatus()).isEqualTo(CheckpointStatsStatus.IN_PROGRESS);
+ assertThat(pending.getNumberOfAcknowledgedSubtasks()).isZero();
+ assertThat(pending.getStateSize()).isZero();
+ assertThat(pending.getNumberOfSubtasks()).isEqualTo(totalSubtaskCount);
+ assertThat(pending.getLatestAcknowledgedSubtaskStats()).isNull();
+ assertThat(pending.getLatestAckTimestamp()).isEqualTo(-1);
+ assertThat(pending.getEndToEndDuration()).isEqualTo(-1);
+ assertThat(pending.getTaskStateStats(task1.getJobVertexId())).isEqualTo(task1);
+ assertThat(pending.getTaskStateStats(task2.getJobVertexId())).isEqualTo(task2);
+ assertThat(pending.getTaskStateStats(new JobVertexID())).isNull();
// Report subtasks and check getters
- assertFalse(pending.reportSubtaskStats(new JobVertexID(), createSubtaskStats(0)));
+ assertThat(pending.reportSubtaskStats(new JobVertexID(), createSubtaskStats(0))).isFalse();
long stateSize = 0;
@@ -83,15 +79,16 @@ public class PendingCheckpointStatsTest {
pending.reportSubtaskStats(task1.getJobVertexId(), subtask);
- assertEquals(subtask, pending.getLatestAcknowledgedSubtaskStats());
- assertEquals(subtask.getAckTimestamp(), pending.getLatestAckTimestamp());
- assertEquals(
- subtask.getAckTimestamp() - triggerTimestamp, pending.getEndToEndDuration());
- assertEquals(stateSize, pending.getStateSize());
+ assertThat(pending.getLatestAcknowledgedSubtaskStats()).isEqualTo(subtask);
+ assertThat(pending.getLatestAckTimestamp()).isEqualTo(subtask.getAckTimestamp());
+ assertThat(pending.getEndToEndDuration())
+ .isEqualTo(subtask.getAckTimestamp() - triggerTimestamp);
+ assertThat(pending.getStateSize()).isEqualTo(stateSize);
}
// Don't allow overwrite
- assertFalse(pending.reportSubtaskStats(task1.getJobVertexId(), task1.getSubtaskStats()[0]));
+ assertThat(pending.reportSubtaskStats(task1.getJobVertexId(), task1.getSubtaskStats()[0]))
+ .isFalse();
// Report 2nd task
for (int i = 0; i < task2.getNumberOfSubtasks(); i++) {
@@ -100,20 +97,20 @@ public class PendingCheckpointStatsTest {
pending.reportSubtaskStats(task2.getJobVertexId(), subtask);
- assertEquals(subtask, pending.getLatestAcknowledgedSubtaskStats());
- assertEquals(subtask.getAckTimestamp(), pending.getLatestAckTimestamp());
- assertEquals(
- subtask.getAckTimestamp() - triggerTimestamp, pending.getEndToEndDuration());
- assertEquals(stateSize, pending.getStateSize());
+ assertThat(pending.getLatestAcknowledgedSubtaskStats()).isEqualTo(subtask);
+ assertThat(pending.getLatestAckTimestamp()).isEqualTo(subtask.getAckTimestamp());
+ assertThat(pending.getEndToEndDuration())
+ .isEqualTo(subtask.getAckTimestamp() - triggerTimestamp);
+ assertThat(pending.getStateSize()).isEqualTo(stateSize);
}
- assertEquals(task1.getNumberOfSubtasks(), task1.getNumberOfAcknowledgedSubtasks());
- assertEquals(task2.getNumberOfSubtasks(), task2.getNumberOfAcknowledgedSubtasks());
+ assertThat(task1.getNumberOfAcknowledgedSubtasks()).isEqualTo(task1.getNumberOfSubtasks());
+ assertThat(task2.getNumberOfAcknowledgedSubtasks()).isEqualTo(task2.getNumberOfSubtasks());
}
/** Test reporting of a completed checkpoint. */
@Test
- public void testReportCompletedCheckpoint() throws Exception {
+ void testReportCompletedCheckpoint() {
TaskStateStats task1 = new TaskStateStats(new JobVertexID(), 3);
TaskStateStats task2 = new TaskStateStats(new JobVertexID(), 4);
@@ -152,30 +149,28 @@ public class PendingCheckpointStatsTest {
CompletedCheckpointStats completed = args.getValue();
- assertNotNull(completed);
- assertEquals(CheckpointStatsStatus.COMPLETED, completed.getStatus());
- assertFalse(completed.isDiscarded());
+ assertThat(completed).isNotNull();
+ assertThat(completed.getStatus()).isEqualTo(CheckpointStatsStatus.COMPLETED);
+ assertThat(completed.isDiscarded()).isFalse();
completed.discard();
- assertTrue(completed.isDiscarded());
- assertEquals(externalPath, completed.getExternalPath());
-
- assertEquals(pending.getCheckpointId(), completed.getCheckpointId());
- assertEquals(
- pending.getNumberOfAcknowledgedSubtasks(),
- completed.getNumberOfAcknowledgedSubtasks());
- assertEquals(
- pending.getLatestAcknowledgedSubtaskStats(),
- completed.getLatestAcknowledgedSubtaskStats());
- assertEquals(pending.getLatestAckTimestamp(), completed.getLatestAckTimestamp());
- assertEquals(pending.getEndToEndDuration(), completed.getEndToEndDuration());
- assertEquals(pending.getStateSize(), completed.getStateSize());
- assertEquals(task1, completed.getTaskStateStats(task1.getJobVertexId()));
- assertEquals(task2, completed.getTaskStateStats(task2.getJobVertexId()));
+ assertThat(completed.isDiscarded()).isTrue();
+ assertThat(completed.getExternalPath()).isEqualTo(externalPath);
+
+ assertThat(completed.getCheckpointId()).isEqualTo(pending.getCheckpointId());
+ assertThat(completed.getNumberOfAcknowledgedSubtasks())
+ .isEqualTo(pending.getNumberOfAcknowledgedSubtasks());
+ assertThat(completed.getLatestAcknowledgedSubtaskStats())
+ .isEqualTo(pending.getLatestAcknowledgedSubtaskStats());
+ assertThat(completed.getLatestAckTimestamp()).isEqualTo(pending.getLatestAckTimestamp());
+ assertThat(completed.getEndToEndDuration()).isEqualTo(pending.getEndToEndDuration());
+ assertThat(completed.getStateSize()).isEqualTo(pending.getStateSize());
+ assertThat(completed.getTaskStateStats(task1.getJobVertexId())).isEqualTo(task1);
+ assertThat(completed.getTaskStateStats(task2.getJobVertexId())).isEqualTo(task2);
}
/** Test reporting of a failed checkpoint. */
@Test
- public void testReportFailedCheckpoint() throws Exception {
+ void testReportFailedCheckpoint() {
TaskStateStats task1 = new TaskStateStats(new JobVertexID(), 3);
TaskStateStats task2 = new TaskStateStats(new JobVertexID(), 4);
@@ -215,27 +210,25 @@ public class PendingCheckpointStatsTest {
FailedCheckpointStats failed = args.getValue();
- assertNotNull(failed);
- assertEquals(CheckpointStatsStatus.FAILED, failed.getStatus());
- assertEquals(failureTimestamp, failed.getFailureTimestamp());
- assertEquals(cause.getMessage(), failed.getFailureMessage());
-
- assertEquals(pending.getCheckpointId(), failed.getCheckpointId());
- assertEquals(
- pending.getNumberOfAcknowledgedSubtasks(),
- failed.getNumberOfAcknowledgedSubtasks());
- assertEquals(
- pending.getLatestAcknowledgedSubtaskStats(),
- failed.getLatestAcknowledgedSubtaskStats());
- assertEquals(pending.getLatestAckTimestamp(), failed.getLatestAckTimestamp());
- assertEquals(failureTimestamp - triggerTimestamp, failed.getEndToEndDuration());
- assertEquals(pending.getStateSize(), failed.getStateSize());
- assertEquals(task1, failed.getTaskStateStats(task1.getJobVertexId()));
- assertEquals(task2, failed.getTaskStateStats(task2.getJobVertexId()));
+ assertThat(failed).isNotNull();
+ assertThat(failed.getStatus()).isEqualTo(CheckpointStatsStatus.FAILED);
+ assertThat(failed.getFailureTimestamp()).isEqualTo(failureTimestamp);
+ assertThat(failed.getFailureMessage()).isEqualTo(cause.getMessage());
+
+ assertThat(failed.getCheckpointId()).isEqualTo(pending.getCheckpointId());
+ assertThat(failed.getNumberOfAcknowledgedSubtasks())
+ .isEqualTo(pending.getNumberOfAcknowledgedSubtasks());
+ assertThat(failed.getLatestAcknowledgedSubtaskStats())
+ .isEqualTo(pending.getLatestAcknowledgedSubtaskStats());
+ assertThat(failed.getLatestAckTimestamp()).isEqualTo(pending.getLatestAckTimestamp());
+ assertThat(failed.getEndToEndDuration()).isEqualTo(failureTimestamp - triggerTimestamp);
+ assertThat(failed.getStateSize()).isEqualTo(pending.getStateSize());
+ assertThat(failed.getTaskStateStats(task1.getJobVertexId())).isEqualTo(task1);
+ assertThat(failed.getTaskStateStats(task2.getJobVertexId())).isEqualTo(task2);
}
@Test
- public void testIsJavaSerializable() throws Exception {
+ void testIsJavaSerializable() throws Exception {
TaskStateStats task1 = new TaskStateStats(new JobVertexID(), 3);
TaskStateStats task2 = new TaskStateStats(new JobVertexID(), 4);
@@ -254,18 +247,17 @@ public class PendingCheckpointStatsTest {
PendingCheckpointStats copy = CommonTestUtils.createCopySerializable(pending);
- assertEquals(pending.getCheckpointId(), copy.getCheckpointId());
- assertEquals(pending.getTriggerTimestamp(), copy.getTriggerTimestamp());
- assertEquals(pending.getProperties(), copy.getProperties());
- assertEquals(pending.getNumberOfSubtasks(), copy.getNumberOfSubtasks());
- assertEquals(
- pending.getNumberOfAcknowledgedSubtasks(), copy.getNumberOfAcknowledgedSubtasks());
- assertEquals(pending.getEndToEndDuration(), copy.getEndToEndDuration());
- assertEquals(pending.getStateSize(), copy.getStateSize());
- assertEquals(
- pending.getLatestAcknowledgedSubtaskStats(),
- copy.getLatestAcknowledgedSubtaskStats());
- assertEquals(pending.getStatus(), copy.getStatus());
+ assertThat(copy.getCheckpointId()).isEqualTo(pending.getCheckpointId());
+ assertThat(copy.getTriggerTimestamp()).isEqualTo(pending.getTriggerTimestamp());
+ assertThat(copy.getProperties()).isEqualTo(pending.getProperties());
+ assertThat(copy.getNumberOfSubtasks()).isEqualTo(pending.getNumberOfSubtasks());
+ assertThat(copy.getNumberOfAcknowledgedSubtasks())
+ .isEqualTo(pending.getNumberOfAcknowledgedSubtasks());
+ assertThat(copy.getEndToEndDuration()).isEqualTo(pending.getEndToEndDuration());
+ assertThat(copy.getStateSize()).isEqualTo(pending.getStateSize());
+ assertThat(copy.getLatestAcknowledgedSubtaskStats())
+ .isEqualTo(pending.getLatestAcknowledgedSubtaskStats());
+ assertThat(copy.getStatus()).isEqualTo(pending.getStatus());
}
// ------------------------------------------------------------------------
[flink] 02/02: [FLINK-31959][checkpoint] Correct the unaligned checkpoint type at checkpoint level
Posted by pn...@apache.org.
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 35254cbdc9a89700a2a99e5b2c7c1a196d7bf2b6
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Thu Apr 27 19:03:58 2023 +0800
[FLINK-31959][checkpoint] Correct the unaligned checkpoint type at checkpoint level
---
.../src/test/resources/rest_api_v1.snapshot | 8 ++--
.../detail/job-checkpoints-detail.component.ts | 8 ++--
.../checkpoint/AbstractCheckpointStats.java | 3 ++
.../checkpoint/CompletedCheckpointStats.java | 12 ++++++
.../runtime/checkpoint/FailedCheckpointStats.java | 3 ++
.../runtime/checkpoint/PendingCheckpointStats.java | 15 +++++++
.../messages/checkpoints/CheckpointStatistics.java | 22 +++++++---
.../CompletedCheckpointStatsSummaryTest.java | 10 ++++-
.../checkpoint/CompletedCheckpointTest.java | 3 ++
.../checkpoint/FailedCheckpointStatsTest.java | 3 ++
.../checkpoint/PendingCheckpointStatsTest.java | 27 ++++++++----
.../checkpoints/CheckpointingStatisticsTest.java | 50 +++++++++++++++++++---
12 files changed, 133 insertions(+), 31 deletions(-)
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index 20d01ddea8a..c2d3597bb71 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -1209,7 +1209,7 @@
},
"checkpoint_type" : {
"type" : "string",
- "enum" : [ "CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
+ "enum" : [ "CHECKPOINT", "UNALIGNED_CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
},
"tasks" : {
"type" : "object",
@@ -1315,7 +1315,7 @@
},
"checkpoint_type" : {
"type" : "string",
- "enum" : [ "CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
+ "enum" : [ "CHECKPOINT", "UNALIGNED_CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
},
"tasks" : {
"type" : "object",
@@ -1403,7 +1403,7 @@
},
"checkpoint_type" : {
"type" : "string",
- "enum" : [ "CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
+ "enum" : [ "CHECKPOINT", "UNALIGNED_CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
},
"tasks" : {
"type" : "object",
@@ -1597,7 +1597,7 @@
},
"checkpoint_type" : {
"type" : "string",
- "enum" : [ "CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
+ "enum" : [ "CHECKPOINT", "UNALIGNED_CHECKPOINT", "SAVEPOINT", "SYNC_SAVEPOINT" ]
},
"tasks" : {
"type" : "object",
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/detail/job-checkpoints-detail.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/detail/job-checkpoints-detail.component.ts
index 9ebf07bff5d..6652c833cd5 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/detail/job-checkpoints-detail.component.ts
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/detail/job-checkpoints-detail.component.ts
@@ -115,11 +115,9 @@ export class JobCheckpointsDetailComponent implements OnInit, OnDestroy {
this.checkPointConfig = config;
this.checkPointDetail = detail;
if (this.checkPointDetail.checkpoint_type === 'CHECKPOINT') {
- if (this.checkPointConfig.unaligned_checkpoints) {
- this.checkPointType = 'unaligned checkpoint';
- } else {
- this.checkPointType = 'aligned checkpoint';
- }
+ this.checkPointType = 'aligned checkpoint';
+ } else if (this.checkPointDetail.checkpoint_type === 'UNALIGNED_CHECKPOINT') {
+ this.checkPointType = 'unaligned checkpoint';
} else if (this.checkPointDetail.checkpoint_type === 'SYNC_SAVEPOINT') {
this.checkPointType = 'savepoint on cancel';
} else if (this.checkPointDetail.checkpoint_type === 'SAVEPOINT') {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractCheckpointStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractCheckpointStats.java
index d27e6c75417..2a3b1d241cb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractCheckpointStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractCheckpointStats.java
@@ -99,6 +99,9 @@ public abstract class AbstractCheckpointStats implements Serializable {
/** @return the total number of persisted bytes during the checkpoint. */
public abstract long getPersistedData();
+ /** @return whether the checkpoint is unaligned. */
+ public abstract boolean isUnalignedCheckpoint();
+
/**
* Returns the latest acknowledged subtask stats or <code>null</code> if none was acknowledged
* yet.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java
index 06cc8a66b9e..1a6888ff0ec 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStats.java
@@ -48,6 +48,8 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
private final long persistedData;
+ private final boolean unalignedCheckpoint;
+
/** The latest acknowledged subtask stats. */
private final SubtaskStateStats latestAcknowledgedSubtask;
@@ -67,6 +69,7 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
long stateSize,
long processedData,
long persistedData,
+ boolean unalignedCheckpoint,
SubtaskStateStats latestAcknowledgedSubtask,
String externalPointer) {
this(
@@ -80,6 +83,7 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
stateSize,
processedData,
persistedData,
+ unalignedCheckpoint,
latestAcknowledgedSubtask,
externalPointer);
}
@@ -98,6 +102,7 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
* @param stateSize Total checkpoint state size over all subtasks.
* @param processedData Processed data during the checkpoint.
* @param persistedData Persisted data during the checkpoint.
+ * @param unalignedCheckpoint Whether the checkpoint is unaligned.
* @param latestAcknowledgedSubtask The latest acknowledged subtask stats.
* @param externalPointer Optional external path if persisted externally.
*/
@@ -112,6 +117,7 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
long stateSize,
long processedData,
long persistedData,
+ boolean unalignedCheckpoint,
SubtaskStateStats latestAcknowledgedSubtask,
String externalPointer) {
@@ -124,6 +130,7 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
this.stateSize = stateSize;
this.processedData = processedData;
this.persistedData = persistedData;
+ this.unalignedCheckpoint = unalignedCheckpoint;
this.latestAcknowledgedSubtask = checkNotNull(latestAcknowledgedSubtask);
this.externalPointer = externalPointer;
}
@@ -158,6 +165,11 @@ public class CompletedCheckpointStats extends AbstractCheckpointStats {
return persistedData;
}
+ @Override
+ public boolean isUnalignedCheckpoint() {
+ return unalignedCheckpoint;
+ }
+
@Override
@Nullable
public SubtaskStateStats getLatestAcknowledgedSubtaskStats() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStats.java
index 43a2a1d0e01..3f63ec22226 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStats.java
@@ -55,6 +55,7 @@ public class FailedCheckpointStats extends PendingCheckpointStats {
* @param stateSize Total checkpoint state size over all subtasks.
* @param processedData Processed data during the checkpoint.
* @param persistedData Persisted data during the checkpoint.
+ * @param unalignedCheckpoint Whether the checkpoint is unaligned.
* @param failureTimestamp Timestamp when this checkpoint failed.
* @param latestAcknowledgedSubtask The latest acknowledged subtask stats or <code>null</code>.
* @param cause Cause of the checkpoint failure or <code>null</code>.
@@ -70,6 +71,7 @@ public class FailedCheckpointStats extends PendingCheckpointStats {
long stateSize,
long processedData,
long persistedData,
+ boolean unalignedCheckpoint,
long failureTimestamp,
@Nullable SubtaskStateStats latestAcknowledgedSubtask,
@Nullable Throwable cause) {
@@ -85,6 +87,7 @@ public class FailedCheckpointStats extends PendingCheckpointStats {
stateSize,
processedData,
persistedData,
+ unalignedCheckpoint,
latestAcknowledgedSubtask);
checkArgument(numAcknowledgedSubtasks >= 0, "Negative number of ACKs");
this.failureTimestamp = failureTimestamp;
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 a8d74f5e581..70cee1ffd42 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
@@ -56,6 +56,8 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
private volatile long currentPersistedData;
+ private volatile boolean unalignedCheckpoint;
+
/** Stats of the latest acknowledged subtask. */
private volatile SubtaskStateStats latestAcknowledgedSubtask;
@@ -110,6 +112,7 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
0,
0,
0,
+ false,
null);
}
@@ -124,6 +127,7 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
long currentStateSize,
long processedData,
long persistedData,
+ boolean unalignedCheckpoint,
@Nullable SubtaskStateStats latestAcknowledgedSubtask) {
super(checkpointId, triggerTimestamp, props, totalSubtaskCount, taskStats);
@@ -131,6 +135,7 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
this.currentStateSize = currentStateSize;
this.currentProcessedData = processedData;
this.currentPersistedData = persistedData;
+ this.unalignedCheckpoint = unalignedCheckpoint;
this.latestAcknowledgedSubtask = latestAcknowledgedSubtask;
this.currentNumAcknowledgedSubtasks = acknowledgedSubtaskCount;
}
@@ -165,6 +170,11 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
return currentPersistedData;
}
+ @Override
+ public boolean isUnalignedCheckpoint() {
+ return unalignedCheckpoint;
+ }
+
@Override
public SubtaskStateStats getLatestAcknowledgedSubtaskStats() {
return latestAcknowledgedSubtask;
@@ -202,6 +212,9 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
if (persistedData > 0) {
currentPersistedData += persistedData;
}
+
+ unalignedCheckpoint |= subtask.getUnalignedCheckpoint();
+
return true;
} else {
return false;
@@ -220,6 +233,7 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
currentStateSize,
currentProcessedData,
currentPersistedData,
+ unalignedCheckpoint,
latestAcknowledgedSubtask,
externalPointer);
}
@@ -242,6 +256,7 @@ public class PendingCheckpointStats extends AbstractCheckpointStats {
currentStateSize,
currentProcessedData,
currentPersistedData,
+ unalignedCheckpoint,
failureTimestamp,
latestAcknowledgedSubtask,
cause);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java
index 2c29cdcaeb9..a8a05ea5f9d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java
@@ -347,7 +347,8 @@ public class CheckpointStatistics implements ResponseBody {
completedCheckpointStats.getNumberOfSubtasks(),
completedCheckpointStats.getNumberOfAcknowledgedSubtasks(),
RestAPICheckpointType.valueOf(
- completedCheckpointStats.getProperties().getCheckpointType()),
+ completedCheckpointStats.getProperties().getCheckpointType(),
+ completedCheckpointStats.isUnalignedCheckpoint()),
checkpointStatisticsPerTask,
completedCheckpointStats.getExternalPath(),
completedCheckpointStats.isDiscarded());
@@ -371,7 +372,8 @@ public class CheckpointStatistics implements ResponseBody {
failedCheckpointStats.getNumberOfSubtasks(),
failedCheckpointStats.getNumberOfAcknowledgedSubtasks(),
RestAPICheckpointType.valueOf(
- failedCheckpointStats.getProperties().getCheckpointType()),
+ failedCheckpointStats.getProperties().getCheckpointType(),
+ failedCheckpointStats.isUnalignedCheckpoint()),
checkpointStatisticsPerTask,
failedCheckpointStats.getFailureTimestamp(),
failedCheckpointStats.getFailureMessage());
@@ -395,7 +397,8 @@ public class CheckpointStatistics implements ResponseBody {
pendingCheckpointStats.getNumberOfSubtasks(),
pendingCheckpointStats.getNumberOfAcknowledgedSubtasks(),
RestAPICheckpointType.valueOf(
- pendingCheckpointStats.getProperties().getCheckpointType()),
+ pendingCheckpointStats.getProperties().getCheckpointType(),
+ pendingCheckpointStats.isUnalignedCheckpoint()),
checkpointStatisticsPerTask);
} else {
throw new IllegalArgumentException(
@@ -411,16 +414,23 @@ public class CheckpointStatistics implements ResponseBody {
*/
enum RestAPICheckpointType {
CHECKPOINT,
+ UNALIGNED_CHECKPOINT,
SAVEPOINT,
SYNC_SAVEPOINT;
- public static RestAPICheckpointType valueOf(SnapshotType checkpointType) {
+ public static RestAPICheckpointType valueOf(
+ SnapshotType checkpointType, boolean isUnalignedCheckpoint) {
if (checkpointType.isSavepoint()) {
+ Preconditions.checkArgument(
+ !isUnalignedCheckpoint,
+ "Currently the savepoint doesn't support unaligned checkpoint.");
SavepointType savepointType = (SavepointType) checkpointType;
return savepointType.isSynchronous() ? SYNC_SAVEPOINT : SAVEPOINT;
- } else {
- return CHECKPOINT;
}
+ if (isUnalignedCheckpoint) {
+ return UNALIGNED_CHECKPOINT;
+ }
+ return CHECKPOINT;
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStatsSummaryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStatsSummaryTest.java
index 4ae0e14163b..ad36a2e955d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStatsSummaryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStatsSummaryTest.java
@@ -42,6 +42,7 @@ public class CompletedCheckpointStatsSummaryTest {
long stateSize = Integer.MAX_VALUE + 17787L;
long processedData = Integer.MAX_VALUE + 123123L;
long persistedData = Integer.MAX_VALUE + 42L;
+ boolean unalignedCheckpoint = true;
CompletedCheckpointStatsSummary summary = new CompletedCheckpointStatsSummary();
assertThat(summary.getStateSizeStats().getCount()).isZero();
@@ -59,7 +60,8 @@ public class CompletedCheckpointStatsSummaryTest {
ackTimestamp + i,
stateSize + i,
processedData + i,
- persistedData + i);
+ persistedData + i,
+ unalignedCheckpoint);
summary.updateSummary(completed);
@@ -93,7 +95,8 @@ public class CompletedCheckpointStatsSummaryTest {
long ackTimestamp,
long stateSize,
long processedData,
- long persistedData) {
+ long persistedData,
+ boolean unalignedCheckpoint) {
SubtaskStateStats latest = mock(SubtaskStateStats.class);
when(latest.getAckTimestamp()).thenReturn(ackTimestamp);
@@ -113,6 +116,7 @@ public class CompletedCheckpointStatsSummaryTest {
stateSize,
processedData,
persistedData,
+ unalignedCheckpoint,
latest,
null);
}
@@ -123,6 +127,7 @@ public class CompletedCheckpointStatsSummaryTest {
int stateSize = 100;
int processedData = 200;
int persistedData = 300;
+ boolean unalignedCheckpoint = true;
long triggerTimestamp = 1234;
long lastAck = triggerTimestamp + 123;
@@ -138,6 +143,7 @@ public class CompletedCheckpointStatsSummaryTest {
stateSize,
processedData,
persistedData,
+ unalignedCheckpoint,
new SubtaskStateStats(0, lastAck),
""));
CompletedCheckpointStatsSummarySnapshot snapshot = summary.createSnapshot();
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 2f6b7259cf2..6170834f684 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
@@ -368,6 +368,7 @@ public class CompletedCheckpointTest {
1,
1,
1,
+ true,
mock(SubtaskStateStats.class),
null);
CompletedCheckpoint completed =
@@ -408,6 +409,7 @@ public class CompletedCheckpointTest {
123129837912L,
42L,
44L,
+ true,
new SubtaskStateStats(
123, 213123, 123123, 123123, 0, 0, 0, 0, 0, 0, false, true),
null);
@@ -424,6 +426,7 @@ public class CompletedCheckpointTest {
assertThat(copy.getStateSize()).isEqualTo(completed.getStateSize());
assertThat(copy.getProcessedData()).isEqualTo(completed.getProcessedData());
assertThat(copy.getPersistedData()).isEqualTo(completed.getPersistedData());
+ assertThat(copy.isUnalignedCheckpoint()).isEqualTo(completed.isUnalignedCheckpoint());
assertThat(copy.getLatestAcknowledgedSubtaskStats().getSubtaskIndex())
.isEqualTo(completed.getLatestAcknowledgedSubtaskStats().getSubtaskIndex());
assertThat(copy.getStatus()).isEqualTo(completed.getStatus());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java
index 28beabc3e05..38cc9c850ba 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailedCheckpointStatsTest.java
@@ -57,6 +57,7 @@ public class FailedCheckpointStatsTest {
0,
0,
0,
+ false,
failureTimestamp,
null,
null);
@@ -87,6 +88,7 @@ public class FailedCheckpointStatsTest {
190890123,
4242,
4444,
+ true,
failureTimestamp,
null,
new NotSerializableException("message"));
@@ -103,6 +105,7 @@ public class FailedCheckpointStatsTest {
assertThat(copy.getStateSize()).isEqualTo(failed.getStateSize());
assertThat(copy.getProcessedData()).isEqualTo(failed.getProcessedData());
assertThat(copy.getPersistedData()).isEqualTo(failed.getPersistedData());
+ assertThat(copy.isUnalignedCheckpoint()).isEqualTo(failed.isUnalignedCheckpoint());
assertThat(copy.getLatestAcknowledgedSubtaskStats())
.isEqualTo(failed.getLatestAcknowledgedSubtaskStats());
assertThat(copy.getStatus()).isEqualTo(failed.getStatus());
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 33d3313c767..65d5a11a0fb 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
@@ -68,17 +68,19 @@ public class PendingCheckpointStatsTest {
assertThat(pending.getTaskStateStats(new JobVertexID())).isNull();
// Report subtasks and check getters
- assertThat(pending.reportSubtaskStats(new JobVertexID(), createSubtaskStats(0))).isFalse();
+ assertThat(pending.reportSubtaskStats(new JobVertexID(), createSubtaskStats(0, false)))
+ .isFalse();
long stateSize = 0;
// Report 1st task
for (int i = 0; i < task1.getNumberOfSubtasks(); i++) {
- SubtaskStateStats subtask = createSubtaskStats(i);
+ SubtaskStateStats subtask = createSubtaskStats(i, false);
stateSize += subtask.getStateSize();
pending.reportSubtaskStats(task1.getJobVertexId(), subtask);
+ assertThat(pending.isUnalignedCheckpoint()).isFalse();
assertThat(pending.getLatestAcknowledgedSubtaskStats()).isEqualTo(subtask);
assertThat(pending.getLatestAckTimestamp()).isEqualTo(subtask.getAckTimestamp());
assertThat(pending.getEndToEndDuration())
@@ -92,11 +94,12 @@ public class PendingCheckpointStatsTest {
// Report 2nd task
for (int i = 0; i < task2.getNumberOfSubtasks(); i++) {
- SubtaskStateStats subtask = createSubtaskStats(i);
+ SubtaskStateStats subtask = createSubtaskStats(i, true);
stateSize += subtask.getStateSize();
pending.reportSubtaskStats(task2.getJobVertexId(), subtask);
+ assertThat(pending.isUnalignedCheckpoint()).isTrue();
assertThat(pending.getLatestAcknowledgedSubtaskStats()).isEqualTo(subtask);
assertThat(pending.getLatestAckTimestamp()).isEqualTo(subtask.getAckTimestamp());
assertThat(pending.getEndToEndDuration())
@@ -131,11 +134,13 @@ public class PendingCheckpointStatsTest {
// Report subtasks
for (int i = 0; i < task1.getNumberOfSubtasks(); i++) {
- pending.reportSubtaskStats(task1.getJobVertexId(), createSubtaskStats(i));
+ pending.reportSubtaskStats(task1.getJobVertexId(), createSubtaskStats(i, false));
+ assertThat(pending.isUnalignedCheckpoint()).isFalse();
}
for (int i = 0; i < task2.getNumberOfSubtasks(); i++) {
- pending.reportSubtaskStats(task2.getJobVertexId(), createSubtaskStats(i));
+ pending.reportSubtaskStats(task2.getJobVertexId(), createSubtaskStats(i, true));
+ assertThat(pending.isUnalignedCheckpoint()).isTrue();
}
// Report completed
@@ -164,6 +169,7 @@ public class PendingCheckpointStatsTest {
assertThat(completed.getLatestAckTimestamp()).isEqualTo(pending.getLatestAckTimestamp());
assertThat(completed.getEndToEndDuration()).isEqualTo(pending.getEndToEndDuration());
assertThat(completed.getStateSize()).isEqualTo(pending.getStateSize());
+ assertThat(completed.isUnalignedCheckpoint()).isTrue();
assertThat(completed.getTaskStateStats(task1.getJobVertexId())).isEqualTo(task1);
assertThat(completed.getTaskStateStats(task2.getJobVertexId())).isEqualTo(task2);
}
@@ -192,11 +198,13 @@ public class PendingCheckpointStatsTest {
// Report subtasks
for (int i = 0; i < task1.getNumberOfSubtasks(); i++) {
- pending.reportSubtaskStats(task1.getJobVertexId(), createSubtaskStats(i));
+ pending.reportSubtaskStats(task1.getJobVertexId(), createSubtaskStats(i, false));
+ assertThat(pending.isUnalignedCheckpoint()).isFalse();
}
for (int i = 0; i < task2.getNumberOfSubtasks(); i++) {
- pending.reportSubtaskStats(task2.getJobVertexId(), createSubtaskStats(i));
+ pending.reportSubtaskStats(task2.getJobVertexId(), createSubtaskStats(i, true));
+ assertThat(pending.isUnalignedCheckpoint()).isTrue();
}
// Report failed
@@ -223,6 +231,7 @@ public class PendingCheckpointStatsTest {
assertThat(failed.getLatestAckTimestamp()).isEqualTo(pending.getLatestAckTimestamp());
assertThat(failed.getEndToEndDuration()).isEqualTo(failureTimestamp - triggerTimestamp);
assertThat(failed.getStateSize()).isEqualTo(pending.getStateSize());
+ assertThat(failed.isUnalignedCheckpoint()).isTrue();
assertThat(failed.getTaskStateStats(task1.getJobVertexId())).isEqualTo(task1);
assertThat(failed.getTaskStateStats(task2.getJobVertexId())).isEqualTo(task2);
}
@@ -262,7 +271,7 @@ public class PendingCheckpointStatsTest {
// ------------------------------------------------------------------------
- private SubtaskStateStats createSubtaskStats(int index) {
+ private SubtaskStateStats createSubtaskStats(int index, boolean unalignedCheckpoint) {
return new SubtaskStateStats(
index,
Integer.MAX_VALUE + (long) index,
@@ -274,7 +283,7 @@ public class PendingCheckpointStatsTest {
Integer.MAX_VALUE + (long) index,
Integer.MAX_VALUE + (long) index,
Integer.MAX_VALUE + (long) index,
- false,
+ unalignedCheckpoint,
true);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatisticsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatisticsTest.java
index e2cab889af9..3afc8daac3a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatisticsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatisticsTest.java
@@ -26,11 +26,16 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.rest.messages.RestResponseMarshallingTestBase;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatistics.RestAPICheckpointType;
+import org.junit.jupiter.api.Test;
+
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
/** Tests for {@link CheckpointingStatistics}. */
public class CheckpointingStatisticsTest
extends RestResponseMarshallingTestBase<CheckpointingStatistics> {
@@ -82,7 +87,7 @@ public class CheckpointingStatisticsTest
44L,
10,
10,
- RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT),
+ RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT, true),
Collections.emptyMap(),
null,
false);
@@ -100,11 +105,11 @@ public class CheckpointingStatisticsTest
1L,
0L,
31337L,
- 4244L,
+ 0L,
9,
9,
RestAPICheckpointType.valueOf(
- SavepointType.savepoint(SavepointFormatType.CANONICAL)),
+ SavepointType.savepoint(SavepointFormatType.CANONICAL), false),
checkpointStatisticsPerTask,
"externalPath",
false);
@@ -125,7 +130,7 @@ public class CheckpointingStatisticsTest
22L,
11,
9,
- RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT),
+ RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT, true),
Collections.emptyMap(),
100L,
"Test failure");
@@ -149,7 +154,7 @@ public class CheckpointingStatisticsTest
16L,
10,
10,
- RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT),
+ RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT, true),
Collections.emptyMap());
final CheckpointingStatistics.LatestCheckpoints latestCheckpoints =
@@ -162,4 +167,39 @@ public class CheckpointingStatisticsTest
latestCheckpoints,
Arrays.asList(completed, savepoint, failed, pending));
}
+
+ @Test
+ void testRestAPICheckpointType() {
+ // Test for aligned checkpoint
+ assertThat(RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT, false))
+ .isEqualTo(RestAPICheckpointType.CHECKPOINT);
+ assertThat(RestAPICheckpointType.valueOf(CheckpointType.FULL_CHECKPOINT, false))
+ .isEqualTo(RestAPICheckpointType.CHECKPOINT);
+
+ // Test for unaligned checkpoint
+ assertThat(RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT, true))
+ .isEqualTo(RestAPICheckpointType.UNALIGNED_CHECKPOINT);
+ assertThat(RestAPICheckpointType.valueOf(CheckpointType.FULL_CHECKPOINT, true))
+ .isEqualTo(RestAPICheckpointType.UNALIGNED_CHECKPOINT);
+
+ // Test for savepoint
+ assertThat(
+ RestAPICheckpointType.valueOf(
+ SavepointType.savepoint(SavepointFormatType.NATIVE), false))
+ .isEqualTo(RestAPICheckpointType.SAVEPOINT);
+ assertThat(
+ RestAPICheckpointType.valueOf(
+ SavepointType.suspend(SavepointFormatType.NATIVE), false))
+ .isEqualTo(RestAPICheckpointType.SYNC_SAVEPOINT);
+ assertThat(
+ RestAPICheckpointType.valueOf(
+ SavepointType.terminate(SavepointFormatType.NATIVE), false))
+ .isEqualTo(RestAPICheckpointType.SYNC_SAVEPOINT);
+
+ assertThatThrownBy(
+ () ->
+ RestAPICheckpointType.valueOf(
+ SavepointType.terminate(SavepointFormatType.NATIVE), true))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
}