[db2] [tests] Fix unstable Db2ConnectorITCase due to cleanup failure

yuxiqian · web-flow · commit 61b9a2ace0a1 · 2023-08-03T16:21:23.000+08:00
This closes #2345.
diff --git a/flink-connector-db2-cdc/src/test/java/com/ververica/cdc/connectors/db2/table/Db2ConnectorITCase.java b/flink-connector-db2-cdc/src/test/java/com/ververica/cdc/connectors/db2/table/Db2ConnectorITCase.java
@@ -62,6 +62,15 @@ public void before() {
         env.setParallelism(1);
     }
 
+    private void cancelJobIfRunning(TableResult result)
+            throws InterruptedException, ExecutionException {
+        try {
+            result.getJobClient().get().cancel().get();
+        } catch (IllegalStateException ignored) {
+            // job isn't running, ignore it
+        }
+    }
+
     @Test
     public void testConsumingAllEvents()
             throws SQLException, InterruptedException, ExecutionException {
@@ -163,7 +172,7 @@ public void testConsumingAllEvents()
         List<String> actual = TestValuesTableFactory.getResults("sink");
         assertThat(actual, containsInAnyOrder(expected));
 
-        result.getJobClient().get().cancel().get();
+        cancelJobIfRunning(result);
     }
 
     @Test
@@ -254,7 +263,7 @@ public void testAllTypes() throws Exception {
         List<String> actual = TestValuesTableFactory.getRawResults("sink");
         assertEquals(expected, actual);
 
-        result.getJobClient().get().cancel().get();
+        cancelJobIfRunning(result);
     }
 
     @Test
@@ -322,7 +331,7 @@ public void testStartupFromLatestOffset() throws Exception {
         List<String> actual = TestValuesTableFactory.getResults("sink");
         assertThat(actual, containsInAnyOrder(expected));
 
-        result.getJobClient().get().cancel().get();
+        cancelJobIfRunning(result);
     }
 
     @Test
@@ -419,7 +428,7 @@ public void testMetadataColumns() throws Throwable {
         Collections.sort(expected);
         Collections.sort(actual);
         assertEquals(expected, actual);
-        result.getJobClient().get().cancel().get();
+        cancelJobIfRunning(result);
     }
 
     private static void waitForSnapshotStarted(String sinkName) throws InterruptedException {
