From 0ce8574603aac784da30a354f608ff939e09ef58 Mon Sep 17 00:00:00 2001
From: zentol <chesnay@apache.org>
Date: Tue, 20 Mar 2018 11:38:24 +0100
Subject: [PATCH] [hotfix][utils] Add ExceptionUtils#findThrowable with
 predicate

---
 .../org/apache/flink/util/ExceptionUtils.java | 25 +++++++++++++++++++
 1 file changed, 25 insertions(+)

diff --git a/flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java b/flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java
index 6af16fcfa4f6c..459648fdc8265 100644
--- a/flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java
@@ -35,6 +35,7 @@
 import java.util.Optional;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
+import java.util.function.Predicate;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -325,6 +326,30 @@ public static <T extends Throwable> Optional<T> findThrowable(Throwable throwabl
 		return Optional.empty();
 	}
 
+	/**
+	 * Checks whether a throwable chain contains an exception matching a predicate and returns it.
+	 *
+	 * @param throwable the throwable chain to check.
+	 * @param predicate the predicate of the exception to search for in the chain.
+	 * @return Optional throwable of the requested type if available, otherwise empty
+	 */
+	public static Optional<Throwable> findThrowable(Throwable throwable, Predicate<Throwable> predicate) {
+		if (throwable == null || predicate == null) {
+			return Optional.empty();
+		}
+
+		Throwable t = throwable;
+		while (t != null) {
+			if (predicate.test(t)) {
+				return Optional.of(t);
+			} else {
+				t = t.getCause();
+			}
+		}
+
+		return Optional.empty();
+	}
+
 	/**
 	 * Checks whether a throwable chain contains a specific error message and returns the corresponding throwable.
 	 *