[hotfix] Add ThrowingRunnable#unchecked and FunctionUtils#uncheckedCo…

…nsumer

ThrowingRunnable#unchecked converts a ThrowingRunnable into a Runnable which throws checked
exceptions as unchecked ones. FunctionUtils#uncheckedConsmer(ThrowingConsumer) converts a
ThrowingConsumer into a Consumer which throws checked exceptions as unchecked ones. This is
necessary because ThrowingConsumer is public and we cannot add new methods to the interface.

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

@@ -20,6 +20,7 @@
 
 import org.apache.flink.util.ExceptionUtils;
 
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 /**
@@ -50,4 +51,22 @@ public static <A, B> Function<A, B> uncheckedFunction(FunctionWithException<A, B
  }
  };
  }
+
+ /**
+ * Converts a {@link ThrowingConsumer} into a {@link Consumer} which throws checked exceptions
+ * as unchecked.
+ *
+ * @param throwingConsumer to convert into a {@link Consumer}
+ * @param <A> input type
+ * @return {@link Consumer} which throws all checked exceptions as unchecked
+ */
+ public static <A> Consumer<A> uncheckedConsumer(ThrowingConsumer<A, ?> throwingConsumer) {
+ return (A value) -> {
+ try {
+ throwingConsumer.accept(value);
+ } catch (Throwable t) {
+ ExceptionUtils.rethrow(t);
+ }
+ };
+ }
 }

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

@@ -19,6 +19,7 @@
 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
@@ -35,4 +36,21 @@ public interface ThrowingRunnable<E extends Throwable> {
  * @throws E Exceptions may be thrown.
  */
  void run() throws E;
+
+ /**
+ * Converts a {@link ThrowingRunnable} into a {@link Runnable} which throws all checked exceptions
+ * as unchecked.
+ *
+ * @param throwingRunnable to convert into a {@link Runnable}
+ * @return {@link Runnable} which throws all checked exceptions as unchecked.
+ */
+ static Runnable unchecked(ThrowingRunnable<?> throwingRunnable) {
+ return () -> {
+ try {
+ throwingRunnable.run();
+ } catch (Throwable t) {
+ ExceptionUtils.rethrow(t);
+ }
+ };
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java

@@ -25,7 +25,7 @@
 import org.apache.flink.runtime.zookeeper.RetrievableStateStorageHelper;
 import org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore;
 import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.function.ConsumerWithException;
+import org.apache.flink.util.function.ThrowingConsumer;
 
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.utils.ZKPaths;
@@ -246,7 +246,7 @@ public void addCheckpoint(final CompletedCheckpoint checkpoint) throws Exception
  LOG.debug("Added {} to {}.", checkpoint, path);
  }
 
- private void tryRemoveCompletedCheckpoint(CompletedCheckpoint completedCheckpoint, ConsumerWithException<CompletedCheckpoint, Exception> discardCallback) {
+ private void tryRemoveCompletedCheckpoint(CompletedCheckpoint completedCheckpoint, ThrowingConsumer<CompletedCheckpoint, Exception> discardCallback) {
  try {
  if (tryRemove(completedCheckpoint.getCheckpointID())) {
  executor.execute(() -> {

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java

@@ -64,8 +64,9 @@
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.function.BiFunctionWithException;
-import org.apache.flink.util.function.ConsumerWithException;
 import org.apache.flink.util.function.FunctionUtils;
+import org.apache.flink.util.function.ThrowingConsumer;
+import org.apache.flink.util.function.ThrowingRunnable;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -849,7 +850,7 @@ private CompletableFuture<Boolean> tryAcceptLeadershipAndRunJobs(UUID newLeaderS
  }
  }
 
- private CompletableFuture<Void> waitForTerminatingJobManager(JobID jobId, JobGraph jobGraph, ConsumerWithException<JobGraph, ?> action) {
+ private CompletableFuture<Void> waitForTerminatingJobManager(JobID jobId, JobGraph jobGraph, ThrowingConsumer<JobGraph, ?> action) {
  final CompletableFuture<Void> jobManagerTerminationFuture = getJobTerminationFuture(jobId)
  .exceptionally((Throwable throwable) -> {
  throw new CompletionException(
@@ -858,10 +859,10 @@ private CompletableFuture<Void> waitForTerminatingJobManager(JobID jobId, JobGra
  throwable)); });
 
  return jobManagerTerminationFuture.thenRunAsync(
- () -> {
+ ThrowingRunnable.unchecked(() -> {
  jobManagerTerminationFutures.remove(jobId);
  action.accept(jobGraph);
- },
+ }),
  getMainThreadExecutor());
  }
 
@@ -934,11 +935,11 @@ public void onAddedJobGraph(final JobID jobId) {
  final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync(
  (Optional<JobGraph> jobGraphOptional) -> jobGraphOptional.map(
  FunctionUtils.uncheckedFunction(jobGraph -> tryRunRecoveredJobGraph(jobGraph, dispatcherId).thenAcceptAsync(
- (ConsumerWithException<Boolean, Exception>) (Boolean isRecoveredJobRunning) -> {
+ FunctionUtils.uncheckedConsumer((Boolean isRecoveredJobRunning) -> {
  if (!isRecoveredJobRunning) {
  submittedJobGraphStore.releaseJobGraph(jobId);
  }
- },
+ }),
  getRpcService().getExecutor())))
  .orElse(CompletableFuture.completedFuture(null)),
  getUnfencedMainThreadExecutor());