[FLINK-37236] Update e2e jar handling logic and improve debugging

Author: gyfora

.github/workflows/e2e.yaml

@@ -84,10 +84,19 @@ jobs:
           if [[ "${{ inputs.append-java-version }}" == "true" ]]; then
             FLINK_IMAGE=${FLINK_IMAGE}-java${{ inputs.java-version }}
           fi
+          
+          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"
+          fi
+          ESCAPED_EXAMPLES_JAR=$(printf '%s\n' "$EXAMPLES_JAR" | sed -e 's/[\/&]/\\&/g')
+          
           echo FLINK_IMAGE=${FLINK_IMAGE}
+          echo EXAMPLES_JAR=${EXAMPLES_JAR}
           sed -i "s/image: flink:.*/image: ${FLINK_IMAGE}/" e2e-tests/data/*.yaml
           sed -i "s/flinkVersion: .*/flinkVersion: ${{ inputs.flink-version }}/" e2e-tests/data/*.yaml
           sed -i "s/mode: .*/mode: ${{ inputs.mode }}/" e2e-tests/data/*.yaml
+          sed -i "s/STREAMING_EXAMPLES_JAR_URL/${ESCAPED_EXAMPLES_JAR}/" e2e-tests/data/*.yaml
           git diff HEAD
           echo "Running e2e-tests/$test"
           bash e2e-tests/${{ inputs.test }} || exit 1

e2e-tests/data/flinkdep-cr.yaml

@@ -31,7 +31,7 @@ spec:
       nginx.ingress.kubernetes.io/rewrite-target: "/$2"
   flinkConfiguration:
     taskmanager.numberOfTaskSlots: "2"
-    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
+    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
@@ -44,7 +44,7 @@ spec:
           image: busybox:1.35.0
           imagePullPolicy: IfNotPresent
           # Use wget or other tools to get user jars from remote storage
-          command: [ 'wget', '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', '-O', '/flink-artifact/myjob.jar' ]
+          command: [ 'wget', 'STREAMING_EXAMPLES_JAR_URL', '-O', '/flink-artifact/myjob.jar' ]
           volumeMounts:
             - mountPath: /flink-artifact
               name: flink-artifact

e2e-tests/data/multi-sessionjob.yaml

@@ -31,7 +31,7 @@ spec:
       nginx.ingress.kubernetes.io/rewrite-target: "/$2"
   flinkConfiguration:
     taskmanager.numberOfTaskSlots: "2"
-    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
+    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
@@ -79,7 +79,7 @@ spec:
       nginx.ingress.kubernetes.io/rewrite-target: "/$2"
   flinkConfiguration:
     taskmanager.numberOfTaskSlots: "2"
-    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
+    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
@@ -120,7 +120,7 @@ metadata:
 spec:
   deploymentName: session-cluster-1
   job:
-    jarURI: 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
+    jarURI: STREAMING_EXAMPLES_JAR_URL
     parallelism: 2
     upgradeMode: savepoint
     entryClass: org.apache.flink.streaming.examples.statemachine.StateMachineExample
@@ -134,7 +134,7 @@ metadata:
 spec:
   deploymentName: session-cluster-1
   job:
-    jarURI: 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
+    jarURI: STREAMING_EXAMPLES_JAR_URL
     parallelism: 2
     upgradeMode: savepoint
     entryClass: org.apache.flink.streaming.examples.statemachine.StateMachineExample

e2e-tests/data/sessionjob-cr.yaml

@@ -31,7 +31,7 @@ spec:
       nginx.ingress.kubernetes.io/rewrite-target: "/$2"
   flinkConfiguration:
     taskmanager.numberOfTaskSlots: "2"
-    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
+    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
@@ -73,7 +73,7 @@ metadata:
 spec:
   deploymentName: session-cluster-1
   job:
-    jarURI: 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
+    jarURI: STREAMING_EXAMPLES_JAR_URL
     parallelism: 2
     upgradeMode: savepoint
     entryClass: org.apache.flink.streaming.examples.statemachine.StateMachineExample

e2e-tests/utils.sh

@@ -231,8 +231,14 @@ function delete_operator_pod_with_leadership() {
 function debug_and_show_logs {
     echo "Debugging failed e2e test:"
     echo "Currently existing Kubernetes resources"
+    kubectl get flinkdeployments
+    kubectl get flinksessionjobs
     kubectl get all
+    kubectl get configmaps
+    kubectl describe flinkdeployments
+    kubectl describe flinksessionjobs
     kubectl describe all
+    kubectl describe configmaps
 
     operator_pod_namespace=$(get_operator_pod_namespace)
     operator_pod_names=$(get_operator_pod_name)

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java

@@ -285,11 +285,13 @@ protected JobUpgrade getJobUpgrade(FlinkResourceContext<CR> ctx, Configuration d
 
             boolean cancellable = allowLastStateCancel(ctx);
             if (running) {
-                return getUpgradeModeBasedOnStateAge(ctx, deployConfig, cancellable);
+                var mode = getUpgradeModeBasedOnStateAge(ctx, deployConfig, cancellable);
+                LOG.info("Job is running, using {} for last-state upgrade", mode);
+                return mode;
             }
 
             if (cancellable) {
-                LOG.info("Using cancel to perform last-state upgrade");
+                LOG.info("Job is not running, using cancel to perform last-state upgrade");
                 return JobUpgrade.lastStateUsingCancel();
             }
         }
@@ -356,8 +358,11 @@ private boolean allowLastStateCancel(FlinkResourceContext<CR> ctx) {
         }
 
         var conf = ctx.getObserveConfig();
-        return conf.get(KubernetesOperatorConfigOptions.OPERATOR_JOB_UPGRADE_LAST_STATE_CANCEL_JOB)
-                || !ctx.getFlinkService().isHaMetadataAvailable(conf);
+        if (!ctx.getFlinkService().isHaMetadataAvailable(conf)) {
+            LOG.info("HA metadata not available, cancel will be used instead of last-state");
+            return true;
+        }
+        return conf.get(KubernetesOperatorConfigOptions.OPERATOR_JOB_UPGRADE_LAST_STATE_CANCEL_JOB);
     }
 
     protected void restoreJob(

helm/flink-kubernetes-operator/conf/log4j-operator.properties

@@ -27,11 +27,11 @@ appender.console.layout.pattern = %style{%d}{yellow} %style{%-30c{1.}}{cyan} %hi
 
 # Do not log config loading
 logger.conf.name = org.apache.flink.configuration.GlobalConfiguration
-logger.conf.level = WARN
+logger.conf.level = ERROR
 
 # Avoid logging fallback key INFO messages
 logger.conf.name = org.apache.flink.configuration.Configuration
-logger.conf.level = WARN
+logger.conf.level = ERROR
 
 # The monitor interval in seconds to enable log4j automatic reconfiguration
 # monitorInterval = 30