[FLINK-35158][Core/API] Move functional interfaces about consumers fr…

…om core module to core-api

flink-core/src/main/java/org/apache/flink/util/function/BiConsumerWithException.java → flink-core-api/src/main/java/org/apache/flink/util/function/BiConsumerWithException.java

@@ -18,8 +18,6 @@
 
 package org.apache.flink.util.function;
 
-import org.apache.flink.util.ExceptionUtils;
-
 import java.util.function.BiConsumer;
 
 /**
@@ -56,7 +54,7 @@ static <A, B> BiConsumer<A, B> unchecked(
             try {
                 biConsumerWithException.accept(a, b);
             } catch (Throwable t) {
-                ExceptionUtils.rethrow(t);
+                ThrowingExceptionUtils.rethrow(t);
             }
         };
     }

flink-core/src/main/java/org/apache/flink/util/function/BiFunctionWithException.java → flink-core-api/src/main/java/org/apache/flink/util/function/BiFunctionWithException.java

@@ -18,8 +18,6 @@
 
 package org.apache.flink.util.function;
 
-import org.apache.flink.util.ExceptionUtils;
-
 import java.util.function.BiFunction;
 
 /**
@@ -58,7 +56,7 @@ static <A, B, C> BiFunction<A, B, C> unchecked(
             try {
                 return biFunctionWithException.apply(a, b);
             } catch (Throwable t) {
-                ExceptionUtils.rethrow(t);
+                ThrowingExceptionUtils.rethrow(t);
                 // we need this to appease the compiler :-(
                 return null;
             }

flink-core/src/main/java/org/apache/flink/util/function/CheckedSupplier.java → flink-core-api/src/main/java/org/apache/flink/util/function/CheckedSupplier.java

@@ -18,8 +18,6 @@
 
 package org.apache.flink.util.function;
 
-import org.apache.flink.util.FlinkException;
-
 import java.util.function.Supplier;
 
 /** Similar to {@link java.util.function.Supplier} but can throw {@link Exception}. */
@@ -41,7 +39,7 @@ static <R> CheckedSupplier<R> checked(Supplier<R> supplier) {
             try {
                 return supplier.get();
             } catch (RuntimeException e) {
-                throw new FlinkException(e);
+                throw new Exception(e);
             }
         };
     }

flink-core-api/src/main/java/org/apache/flink/util/function/ThrowingExceptionUtils.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.util.function;
+
+/** Exception utils only for this package. */
+class ThrowingExceptionUtils {
+
+    /**
+     * Throws the given {@code Throwable} in scenarios where the signatures do not allow you to
+     * throw an arbitrary Throwable. Errors and RuntimeExceptions are thrown directly, other
+     * exceptions are packed into runtime exceptions
+     *
+     * @param t The throwable to be thrown.
+     */
+    static void rethrow(Throwable t) {
+        if (t instanceof Error) {
+            throw (Error) t;
+        } else if (t instanceof RuntimeException) {
+            throw (RuntimeException) t;
+        } else {
+            throw new RuntimeException(t);
+        }
+    }
+
+    private ThrowingExceptionUtils() {}
+}

flink-core/src/main/java/org/apache/flink/util/function/ThrowingRunnable.java → flink-core-api/src/main/java/org/apache/flink/util/function/ThrowingRunnable.java

@@ -19,7 +19,6 @@
 package org.apache.flink.util.function;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.util.ExceptionUtils;
 
 /**
  * Similar to a {@link Runnable}, this interface is used to capture a block of code to be executed.
@@ -48,7 +47,7 @@ static Runnable unchecked(ThrowingRunnable<?> throwingRunnable) {
             try {
                 throwingRunnable.run();
             } catch (Throwable t) {
-                ExceptionUtils.rethrow(t);
+                ThrowingExceptionUtils.rethrow(t);
             }
         };
     }

flink-core/src/main/java/org/apache/flink/util/function/TriConsumerWithException.java → flink-core-api/src/main/java/org/apache/flink/util/function/TriConsumerWithException.java

@@ -18,8 +18,6 @@
 
 package org.apache.flink.util.function;
 
-import org.apache.flink.util.ExceptionUtils;
-
 /**
  * A checked extension of the {@link TriConsumer} interface.
  *
@@ -57,7 +55,7 @@ static <A, B, C> TriConsumer<A, B, C> unchecked(
             try {
                 triConsumerWithException.accept(a, b, c);
             } catch (Throwable t) {
-                ExceptionUtils.rethrow(t);
+                ThrowingExceptionUtils.rethrow(t);
             }
         };
     }

flink-core/src/main/java/org/apache/flink/util/function/TriFunctionWithException.java → flink-core-api/src/main/java/org/apache/flink/util/function/TriFunctionWithException.java

@@ -19,7 +19,6 @@
 package org.apache.flink.util.function;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.util.ExceptionUtils;
 
 import java.util.function.BiFunction;
 
@@ -63,7 +62,7 @@ static <A, B, C, D> TriFunction<A, B, C, D> unchecked(
             try {
                 return triFunctionWithException.apply(a, b, c);
             } catch (Throwable t) {
-                ExceptionUtils.rethrow(t);
+                ThrowingExceptionUtils.rethrow(t);
                 // we need this to appease the compiler :-(
                 return null;
             }

pom.xml

@@ -2365,6 +2365,12 @@ under the License.
 								<!-- MARKER: start exclusions; these will be wiped by tools/releasing/update_japicmp_configuration.sh -->
 								<exclude>org.apache.flink.api.common.functions.Function</exclude>
 								<exclude>org.apache.flink.api.java.functions.KeySelector</exclude>
+								<exclude>org.apache.flink.util.function.FunctionWithException</exclude>
+								<exclude>org.apache.flink.util.function.LongFunctionWithException</exclude>
+								<exclude>org.apache.flink.util.function.RunnableWithException</exclude>
+								<exclude>org.apache.flink.util.function.SerializableFunction</exclude>
+								<exclude>org.apache.flink.util.function.SupplierWithException</exclude>
+								<exclude>org.apache.flink.util.function.ThrowingConsumer</exclude>
 								<!-- MARKER: end exclusions -->
 							</excludes>
 							<accessModifier>public</accessModifier>