[FLINK-21484][rest] Do not expose internal CheckpointType enum via th…

…e REST API Instead use an intermediate class that can provide a backward compatibility and detach the CheckpointType from the REST API.
deepfield · Feb 24, 2021 · fbdc2c0 · fbdc2c0
1 parent 7c286d7
commit fbdc2c0
Show file tree

Hide file tree

Showing 2 changed files with 40 additions and 13 deletions.
diff --git a/...rc/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java b/...rc/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java
@@ -132,7 +132,7 @@ public class CheckpointStatistics implements ResponseBody {
  private final int numAckSubtasks;
 
  @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE)
- private final CheckpointType checkpointType;
+ private final RestAPICheckpointType checkpointType;
 
  @JsonProperty(FIELD_NAME_TASKS)
  @JsonSerialize(keyUsing = JobVertexIDKeySerializer.class)
@@ -152,7 +152,7 @@ private CheckpointStatistics(
  @JsonProperty(FIELD_NAME_PERSISTED_DATA) long persistedData,
  @JsonProperty(FIELD_NAME_NUM_SUBTASKS) int numSubtasks,
  @JsonProperty(FIELD_NAME_NUM_ACK_SUBTASKS) int numAckSubtasks,
- @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE) CheckpointType checkpointType,
+ @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE) RestAPICheckpointType checkpointType,
  @JsonDeserialize(keyUsing = JobVertexIDKeyDeserializer.class)
  @JsonProperty(FIELD_NAME_TASKS)
  Map<JobVertexID, TaskCheckpointStatistics> checkpointStatisticsPerTask) {
@@ -208,7 +208,7 @@ public int getNumAckSubtasks() {
  return numAckSubtasks;
  }
 
- public CheckpointType getCheckpointType() {
+ public RestAPICheckpointType getCheckpointType() {
  return checkpointType;
  }
 
@@ -313,7 +313,8 @@ public static CheckpointStatistics generateCheckpointStatistics(
  completedCheckpointStats.getPersistedData(),
  completedCheckpointStats.getNumberOfSubtasks(),
  completedCheckpointStats.getNumberOfAcknowledgedSubtasks(),
- completedCheckpointStats.getProperties().getCheckpointType(),
+ RestAPICheckpointType.valueOf(
+ completedCheckpointStats.getProperties().getCheckpointType()),
  checkpointStatisticsPerTask,
  completedCheckpointStats.getExternalPath(),
  completedCheckpointStats.isDiscarded());
@@ -334,7 +335,8 @@ public static CheckpointStatistics generateCheckpointStatistics(
  failedCheckpointStats.getPersistedData(),
  failedCheckpointStats.getNumberOfSubtasks(),
  failedCheckpointStats.getNumberOfAcknowledgedSubtasks(),
- failedCheckpointStats.getProperties().getCheckpointType(),
+ RestAPICheckpointType.valueOf(
+ failedCheckpointStats.getProperties().getCheckpointType()),
  checkpointStatisticsPerTask,
  failedCheckpointStats.getFailureTimestamp(),
  failedCheckpointStats.getFailureMessage());
@@ -355,7 +357,8 @@ public static CheckpointStatistics generateCheckpointStatistics(
  pendingCheckpointStats.getPersistedData(),
  pendingCheckpointStats.getNumberOfSubtasks(),
  pendingCheckpointStats.getNumberOfAcknowledgedSubtasks(),
- pendingCheckpointStats.getProperties().getCheckpointType(),
+ RestAPICheckpointType.valueOf(
+ pendingCheckpointStats.getProperties().getCheckpointType()),
  checkpointStatisticsPerTask);
  } else {
  throw new IllegalArgumentException(
@@ -365,6 +368,29 @@ public static CheckpointStatistics generateCheckpointStatistics(
  }
  }
 
+ /**
+ * Backward compatibility layer between internal {@link CheckpointType} and a field used in
+ * {@link CheckpointStatistics}.
+ */
+ enum RestAPICheckpointType {
+ CHECKPOINT,
+ SAVEPOINT,
+ SYNC_SAVEPOINT;
+
+ public static RestAPICheckpointType valueOf(CheckpointType checkpointType) {
+ switch (checkpointType) {
+ case CHECKPOINT:
+ return CHECKPOINT;
+ case SAVEPOINT:
+ return SAVEPOINT;
+ case SYNC_SAVEPOINT:
+ return SYNC_SAVEPOINT;
+ default:
+ throw new UnsupportedOperationException(checkpointType.toString());
+ }
+ }
+ }
+
  // ---------------------------------------------------------------------
  // Static inner classes
  // ---------------------------------------------------------------------
@@ -397,7 +423,7 @@ public CompletedCheckpointStatistics(
  @JsonProperty(FIELD_NAME_PERSISTED_DATA) long persistedData,
  @JsonProperty(FIELD_NAME_NUM_SUBTASKS) int numSubtasks,
  @JsonProperty(FIELD_NAME_NUM_ACK_SUBTASKS) int numAckSubtasks,
- @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE) CheckpointType checkpointType,
+ @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE) RestAPICheckpointType checkpointType,
  @JsonDeserialize(keyUsing = JobVertexIDKeyDeserializer.class)
  @JsonProperty(FIELD_NAME_TASKS)
  Map<JobVertexID, TaskCheckpointStatistics> checkpointingStatisticsPerTask,
@@ -481,7 +507,7 @@ public FailedCheckpointStatistics(
  @JsonProperty(FIELD_NAME_PERSISTED_DATA) long persistedData,
  @JsonProperty(FIELD_NAME_NUM_SUBTASKS) int numSubtasks,
  @JsonProperty(FIELD_NAME_NUM_ACK_SUBTASKS) int numAckSubtasks,
- @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE) CheckpointType checkpointType,
+ @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE) RestAPICheckpointType checkpointType,
  @JsonDeserialize(keyUsing = JobVertexIDKeyDeserializer.class)
  @JsonProperty(FIELD_NAME_TASKS)
  Map<JobVertexID, TaskCheckpointStatistics> checkpointingStatisticsPerTask,
@@ -555,7 +581,7 @@ public PendingCheckpointStatistics(
  @JsonProperty(FIELD_NAME_PERSISTED_DATA) long persistedData,
  @JsonProperty(FIELD_NAME_NUM_SUBTASKS) int numSubtasks,
  @JsonProperty(FIELD_NAME_NUM_ACK_SUBTASKS) int numAckSubtasks,
- @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE) CheckpointType checkpointType,
+ @JsonProperty(FIELD_NAME_CHECKPOINT_TYPE) RestAPICheckpointType checkpointType,
  @JsonDeserialize(keyUsing = JobVertexIDKeyDeserializer.class)
  @JsonProperty(FIELD_NAME_TASKS)
  Map<JobVertexID, TaskCheckpointStatistics> checkpointingStatisticsPerTask) {

diff --git a/.../java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatisticsTest.java b/.../java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatisticsTest.java
@@ -22,6 +22,7 @@
 import org.apache.flink.runtime.checkpoint.CheckpointType;
 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 java.util.Arrays;
 import java.util.Collections;
@@ -76,7 +77,7 @@ protected CheckpointingStatistics getTestResponseInstance() throws Exception {
  44L,
  10,
  10,
- CheckpointType.CHECKPOINT,
+ RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT),
  Collections.emptyMap(),
  null,
  false);
@@ -95,7 +96,7 @@ protected CheckpointingStatistics getTestResponseInstance() throws Exception {
  4244L,
  9,
  9,
- CheckpointType.SAVEPOINT,
+ RestAPICheckpointType.valueOf(CheckpointType.SAVEPOINT),
  checkpointStatisticsPerTask,
  "externalPath",
  false);
@@ -114,7 +115,7 @@ protected CheckpointingStatistics getTestResponseInstance() throws Exception {
  22L,
  11,
  9,
- CheckpointType.CHECKPOINT,
+ RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT),
  Collections.emptyMap(),
  100L,
  "Test failure");
@@ -136,7 +137,7 @@ protected CheckpointingStatistics getTestResponseInstance() throws Exception {
  16L,
  10,
  10,
- CheckpointType.CHECKPOINT,
+ RestAPICheckpointType.valueOf(CheckpointType.CHECKPOINT),
  Collections.emptyMap());
 
  final CheckpointingStatistics.LatestCheckpoints latestCheckpoints =