[FLINK-16056][runtime][tests] do fail ContinuousFileProcessingITCase …

…on failure

flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileProcessingITCase.java

@@ -35,6 +35,7 @@
 import org.apache.flink.streaming.api.functions.source.TimestampedFileInputSplit;
 import org.apache.flink.test.util.AbstractTestBase;
 
+import org.apache.flink.util.ExceptionUtils;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
@@ -53,6 +54,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 
 import static org.junit.Assert.assertEquals;
 
@@ -140,29 +142,19 @@ public void testProgram() throws Exception {
  TestingSinkFunction sink = new TestingSinkFunction();
  content.addSink(sink).setParallelism(1);
 
- Thread job = new Thread() {
-
- @Override
- public void run() {
- try {
- env.execute("ContinuousFileProcessingITCase Job.");
- } catch (Exception e) {
- Throwable th = e;
- for (int depth = 0; depth < 20; depth++) {
- if (th instanceof SuccessException) {
- return;
- } else if (th.getCause() != null) {
- th = th.getCause();
- } else {
- break;
- }
- }
- e.printStackTrace();
- Assert.fail(e.getMessage());
+ CompletableFuture<Void> jobFuture = new CompletableFuture<>();
+ new Thread(() -> {
+ try {
+ env.execute("ContinuousFileProcessingITCase Job.");
+ jobFuture.complete(null);
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, SuccessException.class).isPresent()) {
+ jobFuture.complete(null);
+ } else {
+ jobFuture.completeExceptionally(e);
  }
  }
- };
- job.start();
+ }).start();
 
  // The modification time of the last created file.
  long lastCreatedModTime = Long.MIN_VALUE;
@@ -197,8 +189,7 @@ public void run() {
  Assert.assertTrue(hdfs.exists(file));
  }
 
- // wait for the job to finish.
- job.join();
+ jobFuture.get();
  }
 
  private static class TestingSinkFunction extends RichSinkFunction<String> {