[FLINK-38147] Add Flink 2.1 support

.github/workflows/ci.yml

@@ -91,10 +91,10 @@ jobs:
         http-client: [ "okhttp" ]
         java-version: [ "17"]
         flink-version:
+          - "v2_1"
           - "v2_0"
           - "v1_20"
           - "v1_19"
-          - "v1_18"
     uses: ./.github/workflows/e2e.yaml
     with:
       java-version: ${{ matrix.java-version }}
@@ -108,11 +108,10 @@ jobs:
     strategy:
       matrix:
         flink-version:
+          - "v2_1"
           - "v2_0"
           - "v1_20"
           - "v1_19"
-          - "v1_18"
-          - "v1_17"
         mode:
          - "native"
          - "standalone"
@@ -129,16 +128,6 @@ jobs:
             test: test_dynamic_config.sh
           - mode: standalone
             test: test_dynamic_flink_conf.sh
-          - flink-version: v1_16
-            test: test_autoscaler.sh
-          - flink-version: v1_16
-            test: test_dynamic_config.sh
-          - flink-version: v1_17
-            test: test_dynamic_config.sh
-          - flink-version: v1_18
-            test: test_dynamic_config.sh
-          - flink-version: v1_16
-            test: test_dynamic_flink_conf.sh
 
     uses: ./.github/workflows/e2e.yaml
     with:
@@ -153,11 +142,10 @@ jobs:
     strategy:
       matrix:
         flink-version:
+          - "v2_1"
           - "v2_0"
           - "v1_20"
-          - "v1_18"
           - "v1_19"
-          - "v1_17"
         mode:
           - "native"
           - "standalone"
@@ -173,40 +161,8 @@ jobs:
           - test_snapshot.sh
           - test_batch_job.sh
         exclude:
-          - flink-version: v1_16
-            test: test_autoscaler.sh
           - mode: standalone
             test: test_autoscaler.sh
-          - flink-version: v1_16
-            test: test_dynamic_config.sh
-          - flink-version: v1_16
-            test: test_dynamic_flink_conf.sh
-          - flink-version: v1_16
-            test: test_flink_operator_ha.sh
-          - flink-version: v1_16
-            test: test_snapshot.sh
-          - flink-version: v1_16
-            test: test_batch_job.sh
-          - flink-version: v1_17
-            test: test_dynamic_config.sh
-          - flink-version: v1_17
-            test: test_dynamic_flink_conf.sh
-          - flink-version: v1_17
-            test: test_flink_operator_ha.sh
-          - flink-version: v1_17
-            test: test_snapshot.sh
-          - flink-version: v1_17
-            test: test_batch_job.sh
-          - flink-version: v1_18
-            test: test_dynamic_config.sh
-          - flink-version: v1_18
-            test: test_dynamic_flink_conf.sh
-          - flink-version: v1_18
-            test: test_flink_operator_ha.sh
-          - flink-version: v1_18
-            test: test_snapshot.sh
-          - flink-version: v1_18
-            test: test_batch_job.sh
           - flink-version: v1_19
             test: test_snapshot.sh
     uses: ./.github/workflows/e2e.yaml

.github/workflows/e2e.yaml

@@ -90,7 +90,7 @@ jobs:
 
           EXAMPLES_JAR="https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.14.4/flink-examples-streaming_2.12-1.14.4.jar"
           if [[ ${{ inputs.flink-version }} == v2* ]]; then
-            EXAMPLES_JAR="https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming/2.0-preview1/flink-examples-streaming-2.0-preview1.jar"
+            EXAMPLES_JAR="https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming/2.0.0/flink-examples-streaming-2.0.0.jar"
           elif [[ "${{ inputs.test }}" == "test_batch_job.sh" ]]; then
             EXAMPLES_JAR="https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming/1.20.1/flink-examples-streaming-1.20.1.jar"
           fi

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.16, v1.17, v1.18, v1.19, v1.20
+- Multiple Flink version support: v1.17, v1.18, v1.19, v1.20, v2.0, v2.1
 - [Deployment Modes]({{< ref "docs/custom-resource/overview#application-deployments" >}}):
   - Application cluster
   - Session cluster

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

@@ -118,6 +118,8 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r
 | v1_19 |  |
 | v1_20 |  |
 | v2_0 |  |
+| v2_1 |  |
+| v2_2 |  |
 | majorVersion | int | The major integer from the Flink semver. For example for Flink 1.18.1 this would be 1. |
 | minorVersion | int | The minor integer from the Flink semver. For example for Flink 1.18.1 this would be 18. |
 

e2e-tests/data/savepoint.yaml

@@ -28,5 +28,5 @@ spec:
   savepoint:
     alreadyExists: false
     disposeOnDelete: true
-    formatType: CANONICAL
+    formatType: NATIVE
 

e2e-tests/test_snapshot.sh

@@ -53,7 +53,8 @@ wait_for_status $APPLICATION_IDENTIFIER '.status.jobStatus.state' RUNNING ${TIME
 
 
 # Test upgrade by setting legacy field
-kubectl patch flinkdep ${CLUSTER_ID} --type merge --patch '{"spec":{"job":{"state": "suspended"}}}'
+# Use native savepoints to avoid Flink 2.0/2.1 issues
+kubectl patch flinkdep ${CLUSTER_ID} --type merge --patch '{"spec":{"job":{"state": "suspended"}, "flinkConfiguration":{"kubernetes.operator.savepoint.format.type" : "NATIVE"} }}'
 wait_for_status $APPLICATION_IDENTIFIER '.status.lifecycleState' "SUSPENDED" ${TIMEOUT} || exit 1
 
 location=$(kubectl get $APPLICATION_IDENTIFIER -o yaml | yq '.status.jobStatus.upgradeSavepointPath')

examples/basic-checkpoint-ha.yaml

@@ -29,6 +29,7 @@ spec:
     state.checkpoints.dir: file:///flink-data/checkpoints
     high-availability.type: kubernetes
     high-availability.storageDir: file:///flink-data/ha
+    kubernetes.operator.savepoint.format.type: NATIVE
   serviceAccount: flink
   jobManager:
     resource:

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

@@ -35,11 +35,15 @@ public enum FlinkVersion {
     /** Deprecated since 1.11 operator release. */
     @Deprecated
     v1_16(1, 16),
+    @Deprecated
     v1_17(1, 17),
+    @Deprecated
     v1_18(1, 18),
     v1_19(1, 19),
     v1_20(1, 20),
-    v2_0(2, 0);
+    v2_0(2, 0),
+    v2_1(2, 1),
+    v2_2(2, 2);
 
     /** The major integer from the Flink semver. For example for Flink 1.18.1 this would be 1. */
     private final int majorVersion;

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

@@ -46,6 +46,8 @@ spec:
                 - v1_19
                 - v1_20
                 - v2_0
+                - v2_1
+                - v2_2
                 type: string
               image:
                 type: string