You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2024/08/08 15:00:02 UTC
(flink-kubernetes-operator) 02/04: [FLINK-35265] Remove type field from CheckpointSpec
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit e711ac05db78b0c7bb5cb63f1e6afb248425756a
Author: Mate Czagany <cz...@gmail.com>
AuthorDate: Thu Jul 25 13:52:47 2024 +0200
[FLINK-35265] Remove type field from CheckpointSpec
---
docs/content/docs/custom-resource/reference.md | 4 ++--
.../flink/kubernetes/operator/api/spec/CheckpointSpec.java | 13 +++++--------
.../flink/kubernetes/operator/api/utils/BaseTestUtils.java | 3 +--
.../reconciler/deployment/AbstractJobReconciler.java | 6 +++---
.../reconciler/snapshot/StateSnapshotReconciler.java | 8 ++------
.../kubernetes/operator/utils/FlinkStateSnapshotUtils.java | 7 +------
.../operator/utils/FlinkStateSnapshotUtilsTest.java | 14 ++++----------
.../crds/flinkstatesnapshots.flink.apache.org-v1.yml | 7 -------
8 files changed, 18 insertions(+), 44 deletions(-)
diff --git a/docs/content/docs/custom-resource/reference.md b/docs/content/docs/custom-resource/reference.md
index a385aed3..3bacd6d3 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -51,11 +51,11 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r
### CheckpointSpec
**Class**: org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec
-**Description**: Spec for checkpoint state snapshots.
+**Description**: Spec for checkpoint state snapshots. This is an empty class, used to instruct the operator to
+ trigger a checkpoint.
| Parameter | Type | Docs |
| ----------| ---- | ---- |
-| checkpointType | org.apache.flink.kubernetes.operator.api.status.CheckpointType | Type of checkpoint to take. |
### FlinkDeploymentSpec
**Class**: org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec
diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/CheckpointSpec.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/CheckpointSpec.java
index 7fab2cb8..e25ab52b 100644
--- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/CheckpointSpec.java
+++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/CheckpointSpec.java
@@ -18,22 +18,19 @@
package org.apache.flink.kubernetes.operator.api.spec;
import org.apache.flink.annotation.Experimental;
-import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
-/** Spec for checkpoint state snapshots. */
+/**
+ * Spec for checkpoint state snapshots. This is an empty class, used to instruct the operator to
+ * trigger a checkpoint.
+ */
@Experimental
@Data
@NoArgsConstructor
-@AllArgsConstructor
@Builder
@JsonIgnoreProperties(ignoreUnknown = true)
-public class CheckpointSpec {
- /** Type of checkpoint to take. */
- private CheckpointType checkpointType = CheckpointType.FULL;
-}
+public class CheckpointSpec {}
diff --git a/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/utils/BaseTestUtils.java b/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/utils/BaseTestUtils.java
index 027e32a0..58c5cfe2 100644
--- a/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/utils/BaseTestUtils.java
+++ b/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/utils/BaseTestUtils.java
@@ -249,11 +249,10 @@ public class BaseTestUtils {
String namespace,
CheckpointType checkpointType,
JobReference jobReference) {
- var checkpointSpec = CheckpointSpec.builder().checkpointType(checkpointType).build();
var spec =
FlinkStateSnapshotSpec.builder()
.jobReference(jobReference)
- .checkpoint(checkpointSpec)
+ .checkpoint(new CheckpointSpec())
.build();
var snapshot = new FlinkStateSnapshot();
snapshot.setSpec(spec);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
index 437e925f..565811fd 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
@@ -412,15 +412,15 @@ public abstract class AbstractJobReconciler<
break;
case CHECKPOINT:
- var checkpointType =
- conf.get(KubernetesOperatorConfigOptions.OPERATOR_CHECKPOINT_TYPE);
if (createSnapshotResource) {
FlinkStateSnapshotUtils.createCheckpointResource(
- ctx.getKubernetesClient(), resource, checkpointType, triggerType);
+ ctx.getKubernetesClient(), resource, triggerType);
ReconciliationUtils.updateLastReconciledSnapshotTriggerNonce(
triggerType, resource, CHECKPOINT);
} else {
+ var checkpointType =
+ conf.get(KubernetesOperatorConfigOptions.OPERATOR_CHECKPOINT_TYPE);
var triggerId =
ctx.getFlinkService()
.triggerCheckpoint(
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java
index 216b000c..e37739ce 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java
@@ -18,6 +18,7 @@
package org.apache.flink.kubernetes.operator.reconciler.snapshot;
import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.core.execution.CheckpointType;
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
@@ -216,12 +217,7 @@ public class StateSnapshotReconciler {
throw new IllegalArgumentException(
"Manual checkpoint triggering is not supported for this Flink job (requires Flink 1.17+)");
}
- return Optional.of(
- flinkService.triggerCheckpoint(
- jobId,
- org.apache.flink.core.execution.CheckpointType.valueOf(
- spec.getCheckpoint().getCheckpointType().name()),
- conf));
+ return Optional.of(flinkService.triggerCheckpoint(jobId, CheckpointType.FULL, conf));
} else {
throw new IllegalArgumentException(
"Snapshot must specify either savepoint or checkpoint spec");
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java
index 25191b90..dfe66437 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java
@@ -27,7 +27,6 @@ import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference
import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
import org.apache.flink.kubernetes.operator.api.spec.JobReference;
import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
-import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
@@ -176,21 +175,17 @@ public class FlinkStateSnapshotUtils {
*
* @param kubernetesClient kubernetes client
* @param resource Flink resource associated
- * @param checkpointType type of checkpoint
* @param triggerType trigger type
* @return created snapshot
*/
public static FlinkStateSnapshot createCheckpointResource(
KubernetesClient kubernetesClient,
AbstractFlinkResource<?, ?> resource,
- CheckpointType checkpointType,
SnapshotTriggerType triggerType) {
- var checkpointSpec = CheckpointSpec.builder().checkpointType(checkpointType).build();
-
var snapshotSpec =
FlinkStateSnapshotSpec.builder()
.jobReference(JobReference.fromFlinkResource(resource))
- .checkpoint(checkpointSpec)
+ .checkpoint(new CheckpointSpec())
.build();
var resourceName = getFlinkStateSnapshotName(CHECKPOINT, triggerType, resource);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtilsTest.java
index ec8e9360..7f658d8d 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtilsTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtilsTest.java
@@ -27,7 +27,6 @@ import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference
import org.apache.flink.kubernetes.operator.api.spec.FlinkVersion;
import org.apache.flink.kubernetes.operator.api.spec.JobKind;
import org.apache.flink.kubernetes.operator.api.spec.JobReference;
-import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
@@ -173,14 +172,12 @@ public class FlinkStateSnapshotUtilsTest {
@Test
public void testCreatePeriodicCheckpointResource() {
var deployment = initDeployment();
- var checkpointType = CheckpointType.FULL;
var snapshot =
FlinkStateSnapshotUtils.createCheckpointResource(
- client, deployment, checkpointType, SnapshotTriggerType.PERIODIC);
+ client, deployment, SnapshotTriggerType.PERIODIC);
assertTrue(snapshot.getSpec().isCheckpoint());
- assertEquals(checkpointType, snapshot.getSpec().getCheckpoint().getCheckpointType());
assertEquals(
deployment.getMetadata().getName(), snapshot.getSpec().getJobReference().getName());
assertEquals(JobKind.FLINK_DEPLOYMENT, snapshot.getSpec().getJobReference().getKind());
@@ -238,9 +235,8 @@ public class FlinkStateSnapshotUtilsTest {
var snapshot =
FlinkStateSnapshotUtils.createCheckpointResource(
- client, deployment, CheckpointType.FULL, SnapshotTriggerType.MANUAL);
- assertCheckpointResource(
- snapshot, deployment, SnapshotTriggerType.MANUAL, CheckpointType.FULL);
+ client, deployment, SnapshotTriggerType.MANUAL);
+ assertCheckpointResource(snapshot, deployment, SnapshotTriggerType.MANUAL);
}
@ParameterizedTest
@@ -389,13 +385,11 @@ public class FlinkStateSnapshotUtilsTest {
private void assertCheckpointResource(
FlinkStateSnapshot snapshot,
FlinkDeployment deployment,
- SnapshotTriggerType triggerType,
- CheckpointType checkpointType) {
+ SnapshotTriggerType triggerType) {
assertEquals(
triggerType.name(),
snapshot.getMetadata().getLabels().get(CrdConstants.LABEL_SNAPSHOT_TYPE));
assertTrue(snapshot.getSpec().isCheckpoint());
- assertEquals(checkpointType, snapshot.getSpec().getCheckpoint().getCheckpointType());
assertEquals(
deployment.getMetadata().getName(), snapshot.getSpec().getJobReference().getName());
diff --git a/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml
index 160c64a8..3f2f3475 100644
--- a/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml
+++ b/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml
@@ -38,13 +38,6 @@ spec:
backoffLimit:
type: integer
checkpoint:
- properties:
- checkpointType:
- enum:
- - FULL
- - INCREMENTAL
- - UNKNOWN
- type: string
type: object
jobReference:
properties: