Skip to content

Commit ab333a4

Browse files
AHeiseAHeise
committed
[FLINK-36441] Ignore exceptions during test cleanup
Job may be at various states including race conditions during shutdown. Ideally, the framework would provide idempotence but we can workaround that by ignoring specific exceptions.
1 parent a738104 commit ab333a4

File tree

1 file changed

+58
-29
lines changed

1 file changed

+58
-29
lines changed

‎flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java

Lines changed: 58 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,8 @@
2323
import org.apache.flink.configuration.CoreOptions;
2424
import org.apache.flink.core.execution.JobClient;
2525
import org.apache.flink.core.execution.SavepointFormatType;
26-
import org.apache.flink.core.testutils.FlinkAssertions;
2726
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
27+
import org.apache.flink.runtime.messages.FlinkJobTerminatedWithoutCancellationException;
2828
import org.apache.flink.streaming.api.datastream.DataStream;
2929
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
3030
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
@@ -36,11 +36,14 @@
3636
import org.apache.flink.table.utils.EncodingUtils;
3737
import org.apache.flink.test.util.SuccessException;
3838
import org.apache.flink.types.Row;
39+
import org.apache.flink.util.function.RunnableWithException;
3940

4041
import org.apache.kafka.clients.consumer.NoOffsetForPartitionException;
4142
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
4243
import org.apache.kafka.common.TopicPartition;
44+
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
4345
import org.assertj.core.api.Assertions;
46+
import org.assertj.core.api.ThrowingConsumer;
4447
import org.junit.Before;
4548
import org.junit.Test;
4649
import org.junit.runner.RunWith;
@@ -65,6 +68,7 @@
6568
import java.util.stream.IntStream;
6669

6770
import static org.apache.flink.core.testutils.CommonTestUtils.waitUtil;
71+
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
6872
import static org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestUtils.collectAllRows;
6973
import static org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestUtils.collectRows;
7074
import static org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestUtils.readLines;
@@ -186,7 +190,7 @@ public void testKafkaSourceSink() throws Exception {
186190

187191
// ------------- cleanup -------------------
188192

189-
deleteTestTopic(topic);
193+
cleanupTopic(topic);
190194
}
191195

192196
@Test
@@ -266,8 +270,8 @@ public void testKafkaSourceSinkWithTopicList() throws Exception {
266270
assertThat(topic2Results).containsExactly(Row.of(topic2, 2, 1103, "behavior 2"));
267271

268272
// ------------- cleanup -------------------
269-
deleteTestTopic(topic1);
270-
deleteTestTopic(topic2);
273+
cleanupTopic(topic1);
274+
cleanupTopic(topic2);
271275
}
272276

273277
@Test
@@ -349,8 +353,8 @@ public void testKafkaSourceSinkWithTopicPattern() throws Exception {
349353

350354
// ------------- cleanup -------------------
351355

352-
deleteTestTopic(topic1);
353-
deleteTestTopic(topic2);
356+
cleanupTopic(topic1);
357+
cleanupTopic(topic2);
354358
}
355359

356360
@Test
@@ -406,7 +410,7 @@ public void testKafkaSourceEmptyResultOnDeletedOffsets() throws Exception {
406410

407411
// ------------- cleanup -------------------
408412

409-
deleteTestTopic(topic);
413+
cleanupTopic(topic);
410414
}
411415

412416
@Test
@@ -460,7 +464,7 @@ public void testKafkaSourceSinkWithBoundedSpecificOffsets() throws Exception {
460464

461465
// ------------- cleanup -------------------
462466

463-
deleteTestTopic(topic);
467+
cleanupTopic(topic);
464468
}
465469

466470
@Test
@@ -517,7 +521,7 @@ public void testKafkaSourceSinkWithBoundedTimestamp() throws Exception {
517521

518522
// ------------- cleanup -------------------
519523

520-
deleteTestTopic(topic);
524+
cleanupTopic(topic);
521525
}
522526

523527
@Test
@@ -702,7 +706,7 @@ public void testKafkaSourceSinkWithMetadata() throws Exception {
702706

703707
// ------------- cleanup -------------------
704708

705-
deleteTestTopic(topic);
709+
cleanupTopic(topic);
706710
}
707711

708712
@Test
@@ -783,7 +787,7 @@ public void testKafkaSourceSinkWithKeyAndPartialValue() throws Exception {
783787

784788
// ------------- cleanup -------------------
785789

786-
deleteTestTopic(topic);
790+
cleanupTopic(topic);
787791
}
788792

789793
@Test
@@ -861,7 +865,7 @@ public void testKafkaSourceSinkWithKeyAndFullValue() throws Exception {
861865

862866
// ------------- cleanup -------------------
863867

864-
deleteTestTopic(topic);
868+
cleanupTopic(topic);
865869
}
866870

867871
@Test
@@ -977,8 +981,8 @@ public void testKafkaTemporalJoinChangelog() throws Exception {
977981

978982
// ------------- cleanup -------------------
979983

980-
deleteTestTopic(orderTopic);
981-
deleteTestTopic(productTopic);
984+
cleanupTopic(orderTopic);
985+
cleanupTopic(productTopic);
982986
}
983987

984988
private void initialProductChangelog(String topic, String bootstraps) throws Exception {
@@ -1093,8 +1097,8 @@ public void testPerPartitionWatermarkKafka() throws Exception {
10931097

10941098
// ------------- cleanup -------------------
10951099

1096-
tableResult.getJobClient().ifPresent(JobClient::cancel);
1097-
deleteTestTopic(topic);
1100+
cancelJob(tableResult);
1101+
cleanupTopic(topic);
10981102
}
10991103

11001104
@Test
@@ -1168,8 +1172,8 @@ public void testPerPartitionWatermarkWithIdleSource() throws Exception {
11681172

11691173
// ------------- cleanup -------------------
11701174

1171-
tableResult.getJobClient().ifPresent(JobClient::cancel);
1172-
deleteTestTopic(topic);
1175+
cancelJob(tableResult);
1176+
cleanupTopic(topic);
11731177
}
11741178

11751179
@Test
@@ -1300,8 +1304,8 @@ public void testLatestOffsetStrategyResume() throws Exception {
13001304

13011305
// ------------- cleanup -------------------
13021306

1303-
tableResult.getJobClient().ifPresent(JobClient::cancel);
1304-
deleteTestTopic(topic);
1307+
cancelJob(tableResult);
1308+
cleanupTopic(topic);
13051309
}
13061310

13071311
@Test
@@ -1317,7 +1321,7 @@ public void testStartFromGroupOffsetsEarliest() throws Exception {
13171321
@Test
13181322
public void testStartFromGroupOffsetsNone() {
13191323
Assertions.assertThatThrownBy(() -> testStartFromGroupOffsetsWithNoneResetStrategy())
1320-
.satisfies(FlinkAssertions.anyCauseMatches(NoOffsetForPartitionException.class));
1324+
.satisfies(anyCauseMatches(NoOffsetForPartitionException.class));
13211325
}
13221326

13231327
private List<String> appendNewData(
@@ -1433,10 +1437,8 @@ private void testStartFromGroupOffsets(String resetStrategy) throws Exception {
14331437
KafkaTableTestUtils.waitingExpectedResults(sinkName, expected, Duration.ofSeconds(15));
14341438
} finally {
14351439
// ------------- cleanup -------------------
1436-
if (tableResult != null) {
1437-
tableResult.getJobClient().ifPresent(JobClient::cancel);
1438-
}
1439-
deleteTestTopic(topic);
1440+
cancelJob(tableResult);
1441+
cleanupTopic(topic);
14401442
}
14411443
}
14421444

@@ -1455,10 +1457,8 @@ private void testStartFromGroupOffsetsWithNoneResetStrategy()
14551457
tableResult.await();
14561458
} finally {
14571459
// ------------- cleanup -------------------
1458-
if (tableResult != null) {
1459-
tableResult.getJobClient().ifPresent(JobClient::cancel);
1460-
}
1461-
deleteTestTopic(topic);
1460+
cancelJob(tableResult);
1461+
cleanupTopic(topic);
14621462
}
14631463
}
14641464

@@ -1514,4 +1514,33 @@ private static boolean isCausedByJobFinished(Throwable e) {
15141514
return false;
15151515
}
15161516
}
1517+
1518+
private void cleanupTopic(String topic) {
1519+
ignoreExceptions(
1520+
() -> deleteTestTopic(topic),
1521+
anyCauseMatches(UnknownTopicOrPartitionException.class));
1522+
}
1523+
1524+
private static void cancelJob(TableResult tableResult) {
1525+
if (tableResult != null && tableResult.getJobClient().isPresent()) {
1526+
ignoreExceptions(
1527+
() -> tableResult.getJobClient().get().cancel().get(),
1528+
anyCauseMatches(FlinkJobTerminatedWithoutCancellationException.class),
1529+
anyCauseMatches(
1530+
"MiniCluster is not yet running or has already been shut down."));
1531+
}
1532+
}
1533+
1534+
@SafeVarargs
1535+
private static void ignoreExceptions(
1536+
RunnableWithException runnable, ThrowingConsumer<? super Throwable>... ignoreIf) {
1537+
try {
1538+
runnable.run();
1539+
} catch (InterruptedException e) {
1540+
Thread.currentThread().interrupt();
1541+
} catch (Exception ex) {
1542+
// check if the exception is one of the ignored ones
1543+
assertThat(ex).satisfiesAnyOf(ignoreIf);
1544+
}
1545+
}
15171546
}

0 commit comments

Comments
 (0)