[FLINK-35640] Deprecate Flink version 1.15

mateczagany · gyfora · commit f2adb1593393 · 2024-06-28T15:41:02.000+02:00
diff --git a/docs/content.zh/docs/concepts/overview.md b/docs/content.zh/docs/concepts/overview.md
@@ -44,7 +44,7 @@ Flink Kubernetes Operator 旨在承担人工操作 Flink 部署的职责。 人
   - 有状态和无状态应用程序升级
   - 保存点的触发和管理
   - 处理错误，回滚失败的升级
-- 多 Flink 版本支持：v1.15, v1.16, v1.17, v1.18
+- 多 Flink 版本支持：v1.16, v1.17, v1.18, v1.19
 - [Deployment Modes]({{< ref "docs/custom-resource/overview#application-deployments" >}}):
   - Application 集群
   - Session 集群
diff --git a/docs/content.zh/docs/custom-resource/overview.md b/docs/content.zh/docs/custom-resource/overview.md
@@ -207,7 +207,7 @@ For example, to support the hadoop fs resource:
 ```shell script
 FROM apache/flink-kubernetes-operator
 ENV FLINK_PLUGINS_DIR=/opt/flink/plugins
-COPY flink-hadoop-fs-1.15-SNAPSHOT.jar $FLINK_PLUGINS_DIR/hadoop-fs/
+COPY flink-hadoop-fs-1.19-SNAPSHOT.jar $FLINK_PLUGINS_DIR/hadoop-fs/
 ```
 
 Alternatively, if you use helm to install flink-kubernetes-operator, it allows you to specify a postStart hook to download the required plugins.
diff --git a/docs/content/docs/concepts/overview.md b/docs/content/docs/concepts/overview.md
@@ -36,7 +36,7 @@ Flink Kubernetes Operator aims to capture the responsibilities of a human operat
   - Stateful and stateless application upgrades
   - Triggering and managing savepoints
   - Handling errors, rolling-back broken upgrades
-- Multiple Flink version support: v1.15, v1.16, v1.17, v1.18
+- Multiple Flink version support: v1.16, v1.17, v1.18, v1.19
 - [Deployment Modes]({{< ref "docs/custom-resource/overview#application-deployments" >}}):
   - Application cluster
   - Session cluster
diff --git a/docs/content/docs/custom-resource/overview.md b/docs/content/docs/custom-resource/overview.md
@@ -207,7 +207,7 @@ For example, to support the hadoop fs resource:
 ```shell script
 FROM apache/flink-kubernetes-operator
 ENV FLINK_PLUGINS_DIR=/opt/flink/plugins
-COPY flink-hadoop-fs-1.15-SNAPSHOT.jar $FLINK_PLUGINS_DIR/hadoop-fs/
+COPY flink-hadoop-fs-1.19-SNAPSHOT.jar $FLINK_PLUGINS_DIR/hadoop-fs/
 ```
 
 Alternatively, if you use helm to install flink-kubernetes-operator, it allows you to specify a postStart hook to download the required plugins.
diff --git a/docs/content/docs/custom-resource/reference.md b/docs/content/docs/custom-resource/reference.md
@@ -90,7 +90,7 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r
 | ----- | ---- |
 | v1_13 | No longer supported since 1.7 operator release. |
 | v1_14 | No longer supported since 1.7 operator release. |
-| v1_15 |  |
+| v1_15 | Deprecated since 1.10 operator release. |
 | v1_16 |  |
 | v1_17 |  |
 | v1_18 |  |
diff --git a/examples/flink-beam-example/Dockerfile b/examples/flink-beam-example/Dockerfile
@@ -16,7 +16,7 @@
 # limitations under the License.
 ################################################################################
 
-FROM flink:1.15
+FROM flink:1.16
 
 RUN mkdir /opt/flink/usrlib
 ADD target/flink-beam-example-*.jar /opt/flink/usrlib/beam-runner.jar
diff --git a/examples/flink-beam-example/beam-example.yaml b/examples/flink-beam-example/beam-example.yaml
@@ -22,7 +22,7 @@ metadata:
   name: beam-example
 spec:
   image: flink-beam-example:latest
-  flinkVersion: v1_15
+  flinkVersion: v1_16
   flinkConfiguration:
     taskmanager.numberOfTaskSlots: "1"
   serviceAccount: flink
diff --git a/examples/flink-beam-example/pom.xml b/examples/flink-beam-example/pom.xml
@@ -33,7 +33,7 @@ under the License.
     <!-- Given that this is an example skip maven deployment -->
     <properties>
         <maven.deploy.skip>true</maven.deploy.skip>
-        <beam.version>2.43.0</beam.version>
+        <beam.version>2.47.0</beam.version>
     </properties>
 
     <repositories>
@@ -71,7 +71,7 @@ under the License.
 
         <dependency>
             <groupId>org.apache.beam</groupId>
-            <artifactId>beam-runners-flink-1.15</artifactId>
+            <artifactId>beam-runners-flink-1.16</artifactId>
             <version>${beam.version}</version>
         </dependency>
 
diff --git a/examples/kubernetes-client-examples/src/main/java/org/apache/flink/examples/Basic.java b/examples/kubernetes-client-examples/src/main/java/org/apache/flink/examples/Basic.java
@@ -44,8 +44,8 @@ public static void main(String[] args) {
         objectMeta.setName("basic");
         flinkDeployment.setMetadata(objectMeta);
         FlinkDeploymentSpec flinkDeploymentSpec = new FlinkDeploymentSpec();
-        flinkDeploymentSpec.setFlinkVersion(FlinkVersion.v1_15);
-        flinkDeploymentSpec.setImage("flink:1.15");
+        flinkDeploymentSpec.setFlinkVersion(FlinkVersion.v1_19);
+        flinkDeploymentSpec.setImage("flink:1.19");
         Map<String, String> flinkConfiguration =
                 Map.ofEntries(entry("taskmanager.numberOfTaskSlots", "2"));
         flinkDeploymentSpec.setFlinkConfiguration(flinkConfiguration);
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
@@ -29,6 +29,8 @@ public enum FlinkVersion {
     /** No longer supported since 1.7 operator release. */
     @Deprecated
     v1_14,
+    /** Deprecated since 1.10 operator release. */
+    @Deprecated
     v1_15,
     v1_16,
     v1_17,
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
@@ -325,7 +325,7 @@ public static TestingMetricRegistry createTestMetricRegistry() {
 
     public static Stream<Arguments> flinkVersionsAndUpgradeModes() {
         List<Arguments> args = new ArrayList<>();
-        for (FlinkVersion version : Set.of(FlinkVersion.v1_15, FlinkVersion.v1_18)) {
+        for (FlinkVersion version : Set.of(FlinkVersion.v1_16, FlinkVersion.v1_19)) {
             for (UpgradeMode upgradeMode : UpgradeMode.values()) {
                 args.add(arguments(version, upgradeMode));
             }
@@ -334,7 +334,7 @@ public static Stream<Arguments> flinkVersionsAndUpgradeModes() {
     }
 
     public static Stream<Arguments> flinkVersions() {
-        return Stream.of(arguments(FlinkVersion.v1_15), arguments(FlinkVersion.v1_18));
+        return Stream.of(arguments(FlinkVersion.v1_16), arguments(FlinkVersion.v1_19));
     }
 
     public static FlinkDeployment createCanaryDeployment() {
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
@@ -1058,7 +1058,7 @@ public void testInitialSavepointOnError() throws Exception {
 
     @Test
     public void testInitialHaError() throws Exception {
-        var appCluster = TestUtils.buildApplicationCluster(FlinkVersion.v1_15);
+        var appCluster = TestUtils.buildApplicationCluster(FlinkVersion.v1_19);
         appCluster.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
         testController.reconcile(appCluster, context);
         testController.reconcile(appCluster, context);
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
@@ -693,12 +693,12 @@ public void testRandomJobResultStorePath() throws Exception {
 
     @Test
     public void testAlwaysSavepointOnFlinkVersionChange() throws Exception {
-        var deployment = TestUtils.buildApplicationCluster(FlinkVersion.v1_14);
+        var deployment = TestUtils.buildApplicationCluster(FlinkVersion.v1_18);
         getJobSpec(deployment).setUpgradeMode(UpgradeMode.LAST_STATE);
 
         reconciler.reconcile(deployment, context);
 
-        deployment.getSpec().setFlinkVersion(FlinkVersion.v1_15);
+        deployment.getSpec().setFlinkVersion(FlinkVersion.v1_19);
 
         var reconStatus = deployment.getStatus().getReconciliationStatus();
 
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java
@@ -100,7 +100,7 @@ public class NativeFlinkServiceTest {
     public void setup() {
         configuration.set(KubernetesConfigOptions.CLUSTER_ID, TestUtils.TEST_DEPLOYMENT_NAME);
         configuration.set(KubernetesConfigOptions.NAMESPACE, TestUtils.TEST_NAMESPACE);
-        configuration.set(FLINK_VERSION, FlinkVersion.v1_15);
+        configuration.set(FLINK_VERSION, FlinkVersion.v1_19);
         eventRecorder = new EventRecorder(eventCollector);
         operatorConfig = FlinkOperatorConfiguration.fromConfiguration(configuration);
         executorService = Executors.newDirectExecutorService();
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
@@ -657,15 +657,15 @@ private Consumer<FlinkDeployment> createFlinkVersionChange(
             UpgradeMode fromUpgrade, UpgradeMode toUpgrade, JobState fromState) {
         return dep -> {
             var spec = dep.getSpec();
-            spec.setFlinkVersion(FlinkVersion.v1_15);
+            spec.setFlinkVersion(FlinkVersion.v1_19);
             spec.getJob().setUpgradeMode(toUpgrade);
 
             var suspendSpec = ReconciliationUtils.clone(spec);
 
             // Stopped with LAST_STATE mode with different Flink Version
             suspendSpec.getJob().setUpgradeMode(fromUpgrade);
             suspendSpec.getJob().setState(fromState);
-            suspendSpec.setFlinkVersion(FlinkVersion.v1_14);
+            suspendSpec.setFlinkVersion(FlinkVersion.v1_18);
 
             dep.getStatus()
                     .getReconciliationStatus()
