Addressing Fabian's review

Author: AHeise

flink-connector-kafka/src/test/java/org/apache/flink/KafkaAssertjConfiguration.java

@@ -23,6 +23,8 @@
 /** Configuration for AssertJ that shows full stack traces for unmatched exceptions. */
 public class KafkaAssertjConfiguration extends Configuration {
     public KafkaAssertjConfiguration() {
+        // in case of an assertion error, show the full stack trace
+        // for green builds, this is not changing anything
         setMaxStackTraceElementsDisplayed(10000);
     }
 }

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

@@ -23,7 +23,6 @@
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.core.execution.SavepointFormatType;
-import org.apache.flink.core.testutils.FlinkAssertions;
 import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -36,11 +35,13 @@
 import org.apache.flink.table.utils.EncodingUtils;
 import org.apache.flink.test.util.SuccessException;
 import org.apache.flink.types.Row;
+import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.function.RunnableWithException;
 
 import org.apache.kafka.clients.consumer.NoOffsetForPartitionException;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.assertj.core.api.Assertions;
 import org.junit.Before;
 import org.junit.Test;
@@ -66,6 +67,8 @@
 import java.util.stream.IntStream;
 
 import static org.apache.flink.core.testutils.CommonTestUtils.waitUtil;
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatChainOfCauses;
 import static org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestUtils.collectAllRows;
 import static org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestUtils.collectRows;
 import static org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestUtils.readLines;
@@ -1318,7 +1321,7 @@ public void testStartFromGroupOffsetsEarliest() throws Exception {
     @Test
     public void testStartFromGroupOffsetsNone() {
         Assertions.assertThatThrownBy(() -> testStartFromGroupOffsetsWithNoneResetStrategy())
-                .satisfies(FlinkAssertions.anyCauseMatches(NoOffsetForPartitionException.class));
+                .satisfies(anyCauseMatches(NoOffsetForPartitionException.class));
     }
 
     private List<String> appendNewData(
@@ -1513,20 +1516,28 @@ private static boolean isCausedByJobFinished(Throwable e) {
     }
 
     private void cleanupTopic(String topic) {
-        ignoreExceptions(() -> deleteTestTopic(topic));
+        ignoreExceptions(() -> deleteTestTopic(topic), UnknownTopicOrPartitionException.class);
     }
 
-    private static void ignoreExceptions(RunnableWithException e) {
+    @SafeVarargs
+    private static void ignoreExceptions(
+            RunnableWithException runnable, Class<? extends Exception>... exClasses) {
         try {
-            e.run();
+            runnable.run();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
         } catch (Exception ex) {
-            // ignore
+            // check if the exception is one of the ignored ones
+            assertThatChainOfCauses(ex)
+                    .anyMatch(
+                            cause -> Arrays.stream(exClasses).anyMatch(cl -> cl.isInstance(cause)));
         }
     }
 
     private static void cancelJob(TableResult tableResult) {
-        if (tableResult != null) {
-            ignoreExceptions(() -> tableResult.getJobClient().ifPresent(JobClient::cancel));
+        if (tableResult != null && tableResult.getJobClient().isPresent()) {
+            ignoreExceptions(
+                    () -> tableResult.getJobClient().get().cancel().get(), FlinkException.class);
         }
     }
 }

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/TypeSerializerMatchers.java

@@ -35,7 +35,7 @@
  * A Collection of useful {@link Matcher}s for {@link TypeSerializer} and {@link
  * TypeSerializerSchemaCompatibility}.
  *
- * <p>Note copied from Flink 1.18. Remove when we drop 1.18 support.
+ * <p>Note copied from Flink 1.19. Remove when we drop 1.19 support.
  */
 public final class TypeSerializerMatchers {