You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2024/02/01 15:35:21 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-33089] Clean up 1.13 and 1.14 references from docs and code (#768)
This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 02d8dd5d [FLINK-33089] Clean up 1.13 and 1.14 references from docs and code (#768)
02d8dd5d is described below
commit 02d8dd5d915b7d890d09c23c9003580ab63e6dc1
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Thu Feb 1 16:35:15 2024 +0100
[FLINK-33089] Clean up 1.13 and 1.14 references from docs and code (#768)
---
docs/content/docs/custom-resource/overview.md | 2 +-
docs/content/docs/custom-resource/reference.md | 4 ++--
.../flink/kubernetes/operator/api/spec/FlinkVersion.java | 10 +++++++---
.../flink/kubernetes/operator/config/FlinkConfigBuilder.java | 4 ++--
.../kubernetes/operator/service/AbstractFlinkService.java | 2 +-
.../flink/kubernetes/operator/service/NativeFlinkService.java | 2 +-
.../apache/flink/kubernetes/operator/utils/SnapshotUtils.java | 2 +-
.../operator/controller/FlinkDeploymentControllerTest.java | 2 +-
.../operator/controller/FlinkSessionJobControllerTest.java | 2 +-
.../reconciler/deployment/ApplicationReconcilerTest.java | 2 +-
.../kubernetes/operator/service/AbstractFlinkServiceTest.java | 2 +-
11 files changed, 19 insertions(+), 15 deletions(-)
diff --git a/docs/content/docs/custom-resource/overview.md b/docs/content/docs/custom-resource/overview.md
index 01b098ad..a4362e98 100644
--- a/docs/content/docs/custom-resource/overview.md
+++ b/docs/content/docs/custom-resource/overview.md
@@ -85,7 +85,7 @@ The spec contains all the information the operator need to deploy and manage you
Most deployments will define at least the following fields:
- `image` : Docker used to run Flink job and task manager processes
- - `flinkVersion` : Flink version used in the image (`v1_13`, `v1_14`, `v1_15`, `v1_16` ...)
+ - `flinkVersion` : Flink version used in the image (`v1_15`, `v1_16` ...)
- `serviceAccount` : Kubernetes service account used by the Flink pods
- `taskManager, jobManager` : Job and Task manager pod resource specs (cpu, memory, ephemeralStorage)
- `flinkConfiguration` : Map of Flink configuration overrides such as HA and checkpointing configs
diff --git a/docs/content/docs/custom-resource/reference.md b/docs/content/docs/custom-resource/reference.md
index 030e043a..c7810eb2 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -78,8 +78,8 @@ This page serves as a full reference for FlinkDeployment custom resource definit
| Value | Docs |
| ----- | ---- |
-| v1_13 | |
-| v1_14 | |
+| v1_13 | No longer supported since 1.7 operator release. |
+| v1_14 | No longer supported since 1.7 operator release. |
| v1_15 | |
| v1_16 | |
| v1_17 | |
diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkVersion.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkVersion.java
index f3532b3c..8d4b6b7d 100644
--- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkVersion.java
+++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkVersion.java
@@ -23,7 +23,11 @@ import org.apache.flink.annotation.Experimental;
/** Enumeration for supported Flink versions. */
@Experimental
public enum FlinkVersion {
+ /** No longer supported since 1.7 operator release. */
+ @Deprecated
v1_13,
+ /** No longer supported since 1.7 operator release. */
+ @Deprecated
v1_14,
v1_15,
v1_16,
@@ -31,8 +35,8 @@ public enum FlinkVersion {
v1_18,
v1_19;
- public boolean isNewerVersionThan(FlinkVersion otherVersion) {
- return this.ordinal() > otherVersion.ordinal();
+ public boolean isEqualOrNewer(FlinkVersion otherVersion) {
+ return this.ordinal() >= otherVersion.ordinal();
}
/**
@@ -45,6 +49,6 @@ public enum FlinkVersion {
}
public static boolean isSupported(FlinkVersion version) {
- return version != null && version.isNewerVersionThan(FlinkVersion.v1_14);
+ return version != null && version.isEqualOrNewer(FlinkVersion.v1_15);
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
index 39d0fed6..6caa46a3 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
@@ -118,7 +118,7 @@ public class FlinkConfigBuilder {
protected FlinkConfigBuilder applyImage() {
if (!StringUtils.isNullOrWhitespaceOnly(spec.getImage())) {
String configKey;
- if (spec.getFlinkVersion().isNewerVersionThan(FlinkVersion.v1_16)) {
+ if (spec.getFlinkVersion().isEqualOrNewer(FlinkVersion.v1_17)) {
configKey = KubernetesConfigOptions.CONTAINER_IMAGE.key();
} else {
configKey = "kubernetes.container.image";
@@ -469,7 +469,7 @@ public class FlinkConfigBuilder {
return;
}
- boolean newConfKeys = spec.getFlinkVersion().isNewerVersionThan(FlinkVersion.v1_16);
+ boolean newConfKeys = spec.getFlinkVersion().isEqualOrNewer(FlinkVersion.v1_17);
String configKey;
if (isJM) {
if (newConfKeys) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index 611ab152..2fafb027 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -807,7 +807,7 @@ public abstract class AbstractFlinkService implements FlinkService {
job.getAllowNonRestoredState(),
savepoint,
RestoreMode.DEFAULT,
- conf.get(FLINK_VERSION).isNewerVersionThan(FlinkVersion.v1_16)
+ conf.get(FLINK_VERSION).isEqualOrNewer(FlinkVersion.v1_17)
? conf.toMap()
: null);
LOG.info("Submitting job: {} to session cluster.", jobID);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java
index 41083d91..52bb398e 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java
@@ -250,7 +250,7 @@ public class NativeFlinkService extends AbstractFlinkService {
return false;
}
- if (!observeConfig.get(FLINK_VERSION).isNewerVersionThan(FlinkVersion.v1_17)) {
+ if (!observeConfig.get(FLINK_VERSION).isEqualOrNewer(FlinkVersion.v1_18)) {
LOG.debug("In-place rescaling is only available starting from Flink 1.18");
return false;
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SnapshotUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SnapshotUtils.java
index 1f250b52..2322e484 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SnapshotUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SnapshotUtils.java
@@ -353,7 +353,7 @@ public class SnapshotUtils {
public static boolean isSnapshotTriggeringSupported(Configuration conf) {
// Flink REST API supports triggering checkpoints externally starting with 1.17
return conf.get(FLINK_VERSION) != null
- && conf.get(FLINK_VERSION).isNewerVersionThan(FlinkVersion.v1_16);
+ && conf.get(FLINK_VERSION).isEqualOrNewer(FlinkVersion.v1_17);
}
public static boolean gracePeriodEnded(Duration gracePeriod, SnapshotInfo snapshotInfo) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
index 61dbf692..904944da 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -1231,7 +1231,7 @@ public class FlinkDeploymentControllerTest {
var appCluster = TestUtils.buildApplicationCluster(version);
var updateControl = testController.reconcile(appCluster, context);
var lastEvent = testController.events().poll();
- if (!version.isNewerVersionThan(FlinkVersion.v1_14)) {
+ if (!version.isEqualOrNewer(FlinkVersion.v1_15)) {
assertTrue(updateControl.getScheduleDelay().isEmpty());
assertEquals(
EventRecorder.Reason.UnsupportedFlinkVersion.name(), lastEvent.getReason());
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobControllerTest.java
index 5c620256..d532ff25 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobControllerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobControllerTest.java
@@ -535,7 +535,7 @@ class FlinkSessionJobControllerTest {
Map.of(), kubernetesClient, version);
var updateControl = testController.reconcile(TestUtils.buildSessionJob(), context);
var lastEvent = testController.events().poll();
- if (!version.isNewerVersionThan(FlinkVersion.v1_14)) {
+ if (!version.isEqualOrNewer(FlinkVersion.v1_15)) {
assertTrue(updateControl.getScheduleDelay().isEmpty());
assertEquals(
EventRecorder.Reason.UnsupportedFlinkVersion.name(), lastEvent.getReason());
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index ce83bce5..c414245d 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -946,7 +946,7 @@ public class ApplicationReconcilerTest extends OperatorTestBase {
}
statusRecorder.updateStatusFromCache(deployment);
- if (!flinkVersion.isNewerVersionThan(FlinkVersion.v1_15)) {
+ if (!flinkVersion.isEqualOrNewer(FlinkVersion.v1_16)) {
assertFalse(StringUtils.isBlank(getJobStatus(deployment).getJobId()));
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index 3739298f..637e3bed 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -202,7 +202,7 @@ public class AbstractFlinkServiceTest {
job.getMetadata(), job.getSpec(), JobID.generate(), deployConf, null);
// Make sure that deploy conf was passed to jar run
- if (flinkVersion.isNewerVersionThan(FlinkVersion.v1_16)) {
+ if (flinkVersion.isEqualOrNewer(FlinkVersion.v1_17)) {
assertEquals(deployConf.toMap(), jarRuns.get(0).getFlinkConfiguration().toMap());
} else {
assertTrue(jarRuns.get(0).getFlinkConfiguration().toMap().isEmpty());