tillrohrmann commented on a change in pull request #7568: [FLINK-11417] Make
access to ExecutionGraph single threaded from JobMaster main thread
URL: https://github.com/apache/flink/pull/7568#discussion_r250606022
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/FutureUtils.java
##########
@@ -802,6 +804,89 @@ public void onComplete(Throwable failure, U success) {
return result;
}
+ /**
+ * This function takes a {@link CompletableFuture} and a function to
apply to this future. If the input future
+ * is already done, this function returns {@link
CompletableFuture#thenApply(Function)}. Otherwise, the return
+ * value is {@link CompletableFuture#thenApplyAsync(Function,
Executor)} with the given executor.
+ *
+ * @param completableFuture the completable future for which we want to
apply.
+ * @param executor the executor to run the apply function if the future
is not yet done.
+ * @param applyFun the function to apply.
+ * @param <IN> type of the input future.
+ * @param <OUT> type of the output future.
+ * @return a completable future that is applying the given function to
the input future.
+ */
+ public static <IN, OUT> CompletableFuture<OUT> applyAsyncIfNotDone(
+ CompletableFuture<IN> completableFuture,
+ Executor executor,
+ Function<? super IN, ? extends OUT> applyFun) {
+ return completableFuture.isDone() ?
+ completableFuture.thenApply(applyFun) :
+ completableFuture.thenApplyAsync(applyFun, executor);
+ }
+
+ /**
+ * This function takes a {@link CompletableFuture} and a function to
compose with this future. If the input future
+ * is already done, this function returns {@link
CompletableFuture#thenCompose(Function)}. Otherwise, the return
+ * value is {@link CompletableFuture#thenComposeAsync(Function,
Executor)} with the given executor.
+ *
+ * @param completableFuture the completable future for which we want to
compose.
+ * @param executor the executor to run the compose function if the
future is not yet done.
+ * @param composeFun the function to compose.
+ * @param <IN> type of the input future.
+ * @param <OUT> type of the output future.
+ * @return a completable future that is a composition of the input
future and the function.
+ */
+ public static <IN, OUT> CompletableFuture<OUT> composeAsyncIfNotDone(
+ CompletableFuture<IN> completableFuture,
+ Executor executor,
+ Function<? super IN, ? extends CompletionStage<OUT>>
composeFun) {
Review comment:
This is just personal taste, but it helps to differentiate the parameter
list from the body by indenting the parameters one more level.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
