[FLINK-36129] Supporting Flink 1.20 and removing Flink 1.15

Author: 1996fanrui

.github/workflows/ci.yml

@@ -74,7 +74,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        version: ["v1_19","v1_18","v1_17","v1_16","v1_15"]
+        version: ["v1_20","v1_19","v1_18","v1_17","v1_16"]
         namespace: ["default","flink"]
         mode: ["native", "standalone"]
         java-version: [ 11, 17, 21 ]
@@ -90,6 +90,8 @@ jobs:
         include:
           - namespace: flink
             extraArgs: '--create-namespace --set "watchNamespaces={default,flink}"'
+          - version: v1_20
+            image: flink:1.20
           - version: v1_19
             image: flink:1.19
           - version: v1_18
@@ -98,8 +100,6 @@ jobs:
             image: flink:1.17
           - version: v1_16
             image: flink:1.16
-          - version: v1_15
-            image: flink:1.15
         exclude:
           - namespace: default
             test: test_multi_sessionjob.sh
@@ -111,12 +111,6 @@ jobs:
             test: test_autoscaler.sh
           - mode: standalone
             test: test_dynamic_config.sh
-          - version: v1_15
-            test: test_autoscaler.sh
-          - version: v1_15
-            test: test_dynamic_config.sh
-          - version: v1_15
-            test: test_flink_operator_ha.sh
           - version: v1_16
             test: test_autoscaler.sh
           - version: v1_16
@@ -131,14 +125,10 @@ jobs:
             test: test_dynamic_config.sh
           - version: v1_18
             test: test_flink_operator_ha.sh
-          - version: v1_15
-            java-version: 17
           - version: v1_16
             java-version: 17
           - version: v1_17
             java-version: 17
-          - version: v1_15
-            java-version: 21
           - version: v1_16
             java-version: 21
           - version: v1_17

docs/content.zh/docs/custom-resource/reference.md

@@ -85,6 +85,7 @@ This page serves as a full reference for FlinkDeployment custom resource definit
 | v1_17 |  |
 | v1_18 |  |
 | v1_19 |  |
+| v1_20 |  |
 
 ### IngressSpec
 **Class**: org.apache.flink.kubernetes.operator.api.spec.IngressSpec

docs/content/docs/custom-resource/reference.md

@@ -127,6 +127,7 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r
 | v1_17 |  |
 | v1_18 |  |
 | v1_19 |  |
+| v1_20 |  |
 
 ### IngressSpec
 **Class**: org.apache.flink.kubernetes.operator.api.spec.IngressSpec

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkVersion.java

@@ -35,7 +35,8 @@ public enum FlinkVersion {
     v1_16,
     v1_17,
     v1_18,
-    v1_19;
+    v1_19,
+    v1_20;
 
     public boolean isEqualOrNewer(FlinkVersion otherVersion) {
         return this.ordinal() >= otherVersion.ordinal();

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java

@@ -361,7 +361,7 @@ public static TestingMetricRegistry createTestMetricRegistry() {
 
     public static Stream<Arguments> flinkVersionsAndUpgradeModes() {
         List<Arguments> args = new ArrayList<>();
-        for (FlinkVersion version : Set.of(FlinkVersion.v1_16, FlinkVersion.v1_19)) {
+        for (FlinkVersion version : Set.of(FlinkVersion.v1_16, FlinkVersion.v1_20)) {
             for (UpgradeMode upgradeMode : UpgradeMode.values()) {
                 args.add(arguments(version, upgradeMode));
             }
@@ -370,7 +370,7 @@ public static Stream<Arguments> flinkVersionsAndUpgradeModes() {
     }
 
     public static Stream<Arguments> flinkVersions() {
-        return Stream.of(arguments(FlinkVersion.v1_16), arguments(FlinkVersion.v1_19));
+        return Stream.of(arguments(FlinkVersion.v1_16), arguments(FlinkVersion.v1_20));
     }
 
     public static FlinkDeployment createCanaryDeployment() {

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java

@@ -1063,7 +1063,7 @@ public void testInitialSavepointOnError() throws Exception {
 
     @Test
     public void testInitialHaError() throws Exception {
-        var appCluster = TestUtils.buildApplicationCluster(FlinkVersion.v1_19);
+        var appCluster = TestUtils.buildApplicationCluster(FlinkVersion.v1_20);
         appCluster.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
         testController.reconcile(appCluster, context);
         testController.reconcile(appCluster, context);

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotControllerTest.java

@@ -620,7 +620,7 @@ private FlinkStateSnapshot createCheckpoint(
     }
 
     private FlinkDeployment createDeployment() {
-        return createDeployment(FlinkVersion.v1_19);
+        return createDeployment(FlinkVersion.v1_20);
     }
 
     private FlinkDeployment createDeployment(FlinkVersion flinkVersion) {

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java

@@ -105,7 +105,7 @@ public void setup() {
         configManager = new FlinkConfigManager(configuration);
         configuration.set(KubernetesConfigOptions.CLUSTER_ID, TestUtils.TEST_DEPLOYMENT_NAME);
         configuration.set(KubernetesConfigOptions.NAMESPACE, TestUtils.TEST_NAMESPACE);
-        configuration.set(FLINK_VERSION, FlinkVersion.v1_19);
+        configuration.set(FLINK_VERSION, FlinkVersion.v1_20);
         eventRecorder =
                 new EventRecorder(flinkResourceEventCollector, flinkStateSnapshotEventCollector);
         operatorConfig = FlinkOperatorConfiguration.fromConfiguration(configuration);

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java

@@ -675,7 +675,7 @@ private Consumer<FlinkDeployment> createFlinkVersionChange(
             UpgradeMode fromUpgrade, UpgradeMode toUpgrade, JobState fromState) {
         return dep -> {
             var spec = dep.getSpec();
-            spec.setFlinkVersion(FlinkVersion.v1_19);
+            spec.setFlinkVersion(FlinkVersion.v1_20);
             spec.getJob().setUpgradeMode(toUpgrade);
 
             var suspendSpec = ReconciliationUtils.clone(spec);

helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml

@@ -44,6 +44,7 @@ spec:
                 - v1_17
                 - v1_18
                 - v1_19
+                - v1_20
                 type: string
               image:
                 type: string