[hotfix] Using Flink ExceptionUtils#findThrowable for exception checking (#818)

Update flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java

Co-authored-by: Rui Fan <[email protected]>

* fix

---------

Co-authored-by: Rui Fan <[email protected]>

Author: haoxins

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java

@@ -43,6 +43,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 
+import static org.apache.flink.util.ExceptionUtils.findThrowable;
+
 /** The reconciler for the {@link FlinkSessionJob}. */
 public class SessionJobReconciler
         extends AbstractJobReconciler<FlinkSessionJob, FlinkSessionJobSpec, FlinkSessionJobStatus> {
@@ -127,14 +129,13 @@ public DeleteControl cleanupInternal(FlinkResourceContext<FlinkSessionJob> ctx)
                                     : UpgradeMode.STATELESS;
                     cancelJob(ctx, upgradeMode);
                 } catch (ExecutionException e) {
-                    final var cause = e.getCause();
-
-                    if (cause instanceof FlinkJobNotFoundException) {
+                    if (findThrowable(e, FlinkJobNotFoundException.class).isPresent()) {
                         LOG.error("Job {} not found in the Flink cluster.", jobID, e);
                         return DeleteControl.defaultDelete();
                     }
 
-                    if (cause instanceof FlinkJobTerminatedWithoutCancellationException) {
+                    if (findThrowable(e, FlinkJobTerminatedWithoutCancellationException.class)
+                            .isPresent()) {
                         LOG.error("Job {} already terminated without cancellation.", jobID, e);
                         return DeleteControl.defaultDelete();
                     }

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

@@ -67,9 +67,12 @@
 import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric;
 import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetricsResponseBody;
 import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsHeaders;
+import org.apache.flink.runtime.rest.util.RestClientException;
 import org.apache.flink.util.SerializedThrowable;
 import org.apache.flink.util.concurrent.Executors;
 
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
 import io.fabric8.kubernetes.api.model.DeletionPropagation;
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
@@ -92,6 +95,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeoutException;
@@ -447,7 +451,16 @@ private String cancelJob(FlinkVersion flinkVersion, JobID jobID, boolean savepoi
         }
 
         if (isFlinkJobNotFound) {
-            throw new FlinkJobNotFoundException(jobID);
+            // Throw different exceptions randomly, see
+            // https://github.com/apache/flink-kubernetes-operator/pull/818
+            if (new Random().nextBoolean()) {
+                throw new RestClientException(
+                        "Job could not be found.",
+                        new FlinkJobNotFoundException(jobID),
+                        HttpResponseStatus.NOT_FOUND);
+            } else {
+                throw new FlinkJobNotFoundException(jobID);
+            }
         }
 
         var jobOpt = jobs.stream().filter(js -> js.f1.getJobId().equals(jobID)).findAny();