/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.test.util;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
import org.apache.flink.client.ClientUtils;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.client.JobInitializationException;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.junit.Assert.fail;

/** Test utilities. */
public class TestUtils {

    public static JobExecutionResult tryExecute(StreamExecutionEnvironment see, String name)
            throws Exception {
        try {
            return see.execute(name);
        } catch (ProgramInvocationException | JobExecutionException root) {
            Throwable cause = root.getCause();

            // search for nested SuccessExceptions
            int depth = 0;
            while (!(cause instanceof SuccessException)) {
                if (cause == null || depth++ == 20) {
                    root.printStackTrace();
                    fail("Test failed: " + root.getMessage());
                } else {
                    cause = cause.getCause();
                }
            }
        }

        return null;
    }

    public static void submitJobAndWaitForResult(
            ClusterClient<?> client, JobGraph jobGraph, ClassLoader classLoader) throws Exception {
        client.submitJob(jobGraph)
                .thenCompose(client::requestJobResult)
                .get()
                .toJobExecutionResult(classLoader);
    }

    public static void waitUntilJobInitializationFinished(
            JobID id, MiniClusterWithClientResource miniCluster, ClassLoader userCodeClassloader)
            throws JobInitializationException {
        ClusterClient<?> clusterClient = miniCluster.getClusterClient();
        ClientUtils.waitUntilJobInitializationFinished(
                () -> clusterClient.getJobStatus(id).get(),
                () -> clusterClient.requestJobResult(id).get(),
                userCodeClassloader);
    }
}

@@ -841,63 +841,8 @@
 l;%0A%0A
-import org.apache.flink.api.common.JobExecutionResult;%0A
 impo
@@ -1006,110 +1006,31 @@
 nk.c
-lient.program.ProgramInvocationException;%0Aimport org.apache.flink.runtime.client.JobExecutionException
+ore.execution.JobClient
 ;%0Aim
@@ -1227,132 +1227,479 @@
 nt;%0A
-%0Aimport static org.junit.Assert.fail;%0A%0A/** Test utilities. */%0Apublic class TestUtils %7B%0A%0A    public static JobExecutionResult
+import org.apache.flink.streaming.api.graph.StreamGraph;%0Aimport org.apache.flink.util.ExceptionUtils;%0A%0Aimport java.util.Optional;%0A%0Aimport static org.junit.Assert.fail;%0A%0A/** Test utilities. */%0Apublic class TestUtils %7B%0A%0A    /**%0A     * Execute the job and wait for the job result synchronously.%0A     *%0A     * @throws Exception If executing the environment throws an exception which does not have %7B@link%0A     *     SuccessException%7D as a cause.%0A     */%0A    public static void
  try
@@ -1750,28 +1750,16 @@
 ng name)
-%0A           
  throws 
@@ -1782,44 +1782,100 @@
     
-try %7B%0A            return see.execute
+JobClient jobClient = null;%0A        try %7B%0A            StreamGraph graph = see.getStreamGraph
 (nam
@@ -1890,124 +1890,288 @@
     
+ 
-%7D catch (ProgramInvocationException %7C JobExecutionException root) %7B%0A            Throwable cause = root.getCause();%0A%0A
+   jobClient = see.executeAsync(graph);%0A            jobClient.getJobExecutionResult().get();%0A        %7D catch (Throwable root) %7B%0A            if (jobClient != null) %7B%0A                try %7B%0A                    jobClient.cancel().get();%0A                %7D catch (Exception e) %7B%0A      
     
@@ -2182,45 +2182,115 @@
     
+  
 // 
-search for nested SuccessE
+Exception could be thrown if the job has already finished.%0A                    // Ignore the e
 xception
 s%0A  
@@ -2285,17 +2285,17 @@
 xception
-s
+.
 %0A       
@@ -2303,60 +2303,139 @@
     
-int depth = 0;%0A            while (!(cause instanceof
+    %7D%0A            %7D%0A%0A            Optional%3CSuccessException%3E successAsCause =%0A                    ExceptionUtils.findThrowable(root,
  Suc
@@ -2451,17 +2451,18 @@
 tion
-)) %7B%0A    
+.class);%0A%0A
     
@@ -2477,38 +2477,35 @@
 if (
-cause == null %7C%7C depth++ == 20
+!successAsCause.isPresent()
 ) %7B%0A
@@ -2520,20 +2520,16 @@
         
-    
 root.pri
@@ -2540,28 +2540,24 @@
 ackTrace();%0A
-    
             
@@ -2619,130 +2619,19 @@
     
-    %7D else %7B%0A                    cause = cause.getCause();%0A                %7D%0A            %7D%0A        %7D%0A%0A        return null;
+%7D%0A        %7D
 %0A   

Update TestUtils::tryExecute() to cancel the job after execution failure.