lindong28 commented on a change in pull request #15713:
URL: https://github.com/apache/flink/pull/15713#discussion_r618929637
##########
File path: flink-tests/src/test/java/org/apache/flink/test/util/TestUtils.java
##########
@@ -18,41 +18,50 @@
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.core.execution.JobClient;
import org.apache.flink.runtime.client.JobInitializationException;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.graph.StreamGraph;
import static org.junit.Assert.fail;
/** Test utilities. */
public class TestUtils {
- public static JobExecutionResult tryExecute(StreamExecutionEnvironment
see, String name)
- throws Exception {
+ // Execute the job and wait for the job result synchronously. The method
throws exception
+ // iff one of the following conditions happens:
+ // 1) The job finishes successfully without exception
+ // 2) The job finishes with an exception that contains SuccessException.
+ public static void tryExecute(StreamExecutionEnvironment see, String name)
throws Exception {
+ JobClient jobClient = null;
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();
+ StreamGraph graph = see.getStreamGraph(name);
+ jobClient = see.executeAsync(graph);
+ jobClient.getJobExecutionResult().get();
+ } catch (Throwable root) {
+ if (jobClient != null) {
+ try {
+ jobClient.cancel().get();
+ } catch (Exception e) {
+ // Exception could be thrown if the job has already
finished.
+ // Ignore the exception.
}
}
- }
- return null;
+ Throwable t = root;
Review comment:
Nice!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
