[hotfix][tests] Add matcher for finding exception cause by Class

flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkMatchers.java

@@ -81,6 +81,11 @@ public static <T> FutureWillFailMatcher<T> futureWillCompleteExceptionally(Durat
  return futureWillCompleteExceptionally(Throwable.class, timeout);
  }
 
+ /** Checks for a {@link Throwable} that matches by class. */
+ public static Matcher<Throwable> containsCause(Class<? extends Throwable> failureCause) {
+ return new ContainsCauseMatcher(failureCause);
+ }
+
  /** Checks for a {@link Throwable} that matches by class and message. */
  public static Matcher<Throwable> containsCause(Throwable failureCause) {
  return new ContainsCauseAndMessageMatcher(failureCause);
@@ -229,6 +234,38 @@ public void describeTo(Description description) {
  }
  }
 
+ private static final class ContainsCauseMatcher extends TypeSafeDiagnosingMatcher<Throwable> {
+
+ private final Class<? extends Throwable> failureCause;
+
+ private ContainsCauseMatcher(Class<? extends Throwable> failureCause) {
+ this.failureCause = failureCause;
+ }
+
+ @Override
+ protected boolean matchesSafely(Throwable throwable, Description description) {
+ final Optional<Throwable> optionalCause =
+ findThrowable(throwable, cause -> cause.getClass() == failureCause);
+
+ if (!optionalCause.isPresent()) {
+ description
+ .appendText("The throwable ")
+ .appendValue(throwable)
+ .appendText(" does not contain the expected failure cause ")
+ .appendValue(failureCause.getSimpleName());
+ }
+
+ return optionalCause.isPresent();
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description
+ .appendText("Expected failure cause is ")
+ .appendValue(failureCause.getSimpleName());
+ }
+ }
+
  private static final class ContainsCauseAndMessageMatcher
  extends TypeSafeDiagnosingMatcher<Throwable> {