Tweaks to e2e test

Author: luca-p-castelli

.github/workflows/e2e.yaml

@@ -88,6 +88,8 @@ 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"
+          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
           ESCAPED_EXAMPLES_JAR=$(printf '%s\n' "$EXAMPLES_JAR" | sed -e 's/[\/&]/\\&/g')
           

e2e-tests/data/flinkdep-batch-cr.yaml

@@ -31,10 +31,6 @@ spec:
       nginx.ingress.kubernetes.io/rewrite-target: "/$2"
   flinkConfiguration:
     taskmanager.numberOfTaskSlots: "2"
-    high-availability.type: kubernetes
-    high-availability.storageDir: file:///opt/flink/volume/flink-ha
-    state.checkpoints.dir: file:///opt/flink/volume/flink-cp
-    state.savepoints.dir: file:///opt/flink/volume/flink-sp
     kubernetes.operator.snapshot.resource.enabled: "false"
   serviceAccount: flink
   podTemplate:
@@ -59,14 +55,9 @@ spec:
           volumeMounts:
             - mountPath: /opt/flink/usrlib
               name: flink-artifact
-            - mountPath: /opt/flink/volume
-              name: flink-volume
       volumes:
         - name: flink-artifact
           emptyDir: { }
-        - name: flink-volume
-          persistentVolumeClaim:
-            claimName: flink-example-wordcount-batch
   jobManager:
     resource:
       memory: "1024m"
@@ -83,19 +74,6 @@ spec:
     upgradeMode: stateless
   mode: native
 
----
-apiVersion: v1
-kind: PersistentVolumeClaim
-metadata:
-  name: flink-example-wordcount-batch
-spec:
-  accessModes:
-    - ReadWriteOnce
-  volumeMode: Filesystem
-  resources:
-    requests:
-      storage: 1Gi
-
 ---
 apiVersion: networking.k8s.io/v1
 kind: IngressClass

e2e-tests/test_batch_job.sh

@@ -25,6 +25,7 @@
 # 5. Verifies that the job remains in the FINISHED state (second check).
 # 6. Checks the operator logs for the expected job state transition message.
 # 7. Checks the JobManager logs for successful application completion.
+# 8. Applies a spec change to the FlinkDeployment and verifies the job re-runs successfully.
 SCRIPT_DIR=$(dirname "$(readlink -f "$0")")
 source "${SCRIPT_DIR}/utils.sh"
 
@@ -57,4 +58,9 @@ wait_for_operator_logs "$operator_pod_name" "Job status changed from .* to FINIS
 jm_pod_name=$(get_jm_pod_name $CLUSTER_ID)
 wait_for_logs "$jm_pod_name" "Application completed SUCCESSFULLY" ${TIMEOUT} || exit 1
 
+# Apply a spec change; verify the job re-runs and reaches the FINISHED state.
+kubectl patch flinkdep ${CLUSTER_ID} --type merge --patch '{"spec":{"job": {"parallelism": 1 } } }'
+wait_for_status $APPLICATION_IDENTIFIER '.status.jobStatus.state' RECONCILING $TIMEOUT || exit 1
+wait_for_status $APPLICATION_IDENTIFIER '.status.jobStatus.state' FINISHED $TIMEOUT || exit 1
+
 echo "Successfully ran the batch job test"