Package org.apache.flink.core.asyncprocessing

Class AsyncFutureImpl<T>

java.lang.Object

org.apache.flink.core.asyncprocessing.AsyncFutureImpl<T>

All Implemented Interfaces:

org.apache.flink.api.common.state.v2.StateFuture<T>, AsyncFuture<T>, InternalAsyncFuture<T>

@Internal
public class AsyncFutureImpl<T>
extends Object
implements InternalAsyncFuture<T>

The default implementation of AsyncFuture. This is managed by the runtime framework and should never be directly created in user code. It will handle the completion and callback trigger, and most of the design are borrowed from the CompletableFuture. In the basic version of this implementation, we wrap CompletableFuture for simplification. TODO: remove CompletableFuture.

This class is on hot path and very complex, please take care of the performance as well as the running thread of each block when you decide to touch it.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	AsyncFutureImpl.AsyncFrameworkExceptionHandler	Handle exceptions thrown by async state callback framework.
static interface	AsyncFutureImpl.CallbackRunner	The entry for a state future to submit task to mailbox.

Field Summary

Fields

Modifier and Type	Field	Description
protected AsyncFutureImpl.CallbackRunner	callbackRunner	The callback runner.
protected AsyncFutureImpl.AsyncFrameworkExceptionHandler	exceptionHandler	The exception handler that handles callback framework's error.

Constructor Summary

Constructors

Constructor	Description
AsyncFutureImpl(AsyncFutureImpl.CallbackRunner callbackRunner, AsyncFutureImpl.AsyncFrameworkExceptionHandler exceptionHandler)

Method Summary

Modifier and Type	Method	Description
void	callbackFinished()	Will be triggered when a callback finishes processing.
void	callbackRegistered()	Will be triggered when a callback is registered.
void	complete(T result)	Complete this future.
void	completeExceptionally(String message, Throwable ex)	Fail this future and pass the given exception to the runtime.
T	get()	Waits if necessary for the computation to complete, and then retrieves its result.
boolean	isDone()	Returns true if completed in any fashion: normally, exceptionally, or via cancellation.
<A> AsyncFutureImpl<A>	makeNewFuture()	Make a new future based on context of this future.
void	postComplete(boolean inCallbackRunner)	Will be triggered when this future completes.
InternalAsyncFuture<Void>	thenAccept(org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> action)	Returns a new StateFuture that, when this future completes normally, is executed with this future's result as the argument to the supplied action.
<U> InternalAsyncFuture<U>	thenApply(org.apache.flink.util.function.FunctionWithException<? super T,? extends U,? extends Exception> fn)	Returns a new StateFuture that, when this future completes normally, is executed with this future's result as the argument to the supplied function.
<U,V> InternalAsyncFuture<V>	thenCombine(org.apache.flink.api.common.state.v2.StateFuture<? extends U> other, org.apache.flink.util.function.BiFunctionWithException<? super T,? super U,? extends V,? extends Exception> fn)	Returns a new StateFuture that, when this and the other given future both complete normally, is executed with the two results as arguments to the supplied function.
<U> InternalAsyncFuture<U>	thenCompose(org.apache.flink.util.function.FunctionWithException<? super T,? extends org.apache.flink.api.common.state.v2.StateFuture<U>,? extends Exception> action)	Returns a new future that, when this future completes normally, is executed with this future as the argument to the supplied function.
InternalAsyncFuture<Boolean>	thenConditionallyAccept(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition, org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> actionIfTrue)	Apply a condition test on the result of this StateFuture, and try to perform the action if test result is true.
InternalAsyncFuture<Boolean>	thenConditionallyAccept(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition, org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> actionIfTrue, org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> actionIfFalse)	Apply a condition test on the result of this StateFuture, and try to perform one action out of two based on the result.
<U> InternalAsyncFuture<org.apache.flink.api.java.tuple.Tuple2<Boolean,U>>	thenConditionallyApply(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition, org.apache.flink.util.function.FunctionWithException<? super T,? extends U,? extends Exception> actionIfTrue)	Apply a condition test on the result of this StateFuture, and try to perform the action if test result is true.
<U,V> InternalAsyncFuture<org.apache.flink.api.java.tuple.Tuple2<Boolean,Object>>	thenConditionallyApply(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition, org.apache.flink.util.function.FunctionWithException<? super T,? extends U,? extends Exception> actionIfTrue, org.apache.flink.util.function.FunctionWithException<? super T,? extends V,? extends Exception> actionIfFalse)	Apply a condition test on the result of this StateFuture, and try to perform one action out of two based on the result.
<U> InternalAsyncFuture<org.apache.flink.api.java.tuple.Tuple2<Boolean,U>>	thenConditionallyCompose(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition, org.apache.flink.util.function.FunctionWithException<? super T,? extends org.apache.flink.api.common.state.v2.StateFuture<U>,? extends Exception> actionIfTrue)	Apply a condition test on the result of this StateFuture, and try to perform the action if test result is true.
<U,V> InternalAsyncFuture<org.apache.flink.api.java.tuple.Tuple2<Boolean,Object>>	thenConditionallyCompose(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition, org.apache.flink.util.function.FunctionWithException<? super T,? extends org.apache.flink.api.common.state.v2.StateFuture<U>,? extends Exception> actionIfTrue, org.apache.flink.util.function.FunctionWithException<? super T,? extends org.apache.flink.api.common.state.v2.StateFuture<V>,? extends Exception> actionIfFalse)	Apply a condition test on the result of this StateFuture, and try to perform one action out of two based on the result.
void	thenSyncAccept(org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> action)	Accept the action in the same thread with the one of complete (or current thread if it has been completed).

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

callbackRunner

protected final AsyncFutureImpl.CallbackRunner callbackRunner

The callback runner.

exceptionHandler

protected final AsyncFutureImpl.AsyncFrameworkExceptionHandler exceptionHandler

The exception handler that handles callback framework's error.

Constructor Detail

AsyncFutureImpl

public AsyncFutureImpl(AsyncFutureImpl.CallbackRunner callbackRunner,
                       AsyncFutureImpl.AsyncFrameworkExceptionHandler exceptionHandler)

Method Detail

thenApply

public <U> InternalAsyncFuture<U> thenApply(org.apache.flink.util.function.FunctionWithException<? super T,? extends U,? extends Exception> fn)

Description copied from interface: InternalAsyncFuture

Returns a new StateFuture that, when this future completes normally, is executed with this future's result as the argument to the supplied function.

Specified by:

thenApply in interface AsyncFuture<T>

Specified by:

thenApply in interface InternalAsyncFuture<T>

Specified by:

thenApply in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Type Parameters:

U - the function's return type.

Parameters:

fn - the function to use to compute the value of the returned StateFuture.

Returns:

the new StateFuture.

thenAccept

public InternalAsyncFuture<Void> thenAccept(org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> action)

Description copied from interface: InternalAsyncFuture

Returns a new StateFuture that, when this future completes normally, is executed with this future's result as the argument to the supplied action.

Specified by:

thenAccept in interface AsyncFuture<T>

Specified by:

thenAccept in interface InternalAsyncFuture<T>

Specified by:

thenAccept in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Parameters:

action - the action to perform before completing the returned StateFuture.

Returns:

the new StateFuture.

thenCompose

public <U> InternalAsyncFuture<U> thenCompose(org.apache.flink.util.function.FunctionWithException<? super T,? extends org.apache.flink.api.common.state.v2.StateFuture<U>,? extends Exception> action)

Description copied from interface: InternalAsyncFuture

Returns a new future that, when this future completes normally, is executed with this future as the argument to the supplied function.

Specified by:

thenCompose in interface AsyncFuture<T>

Specified by:

thenCompose in interface InternalAsyncFuture<T>

Specified by:

thenCompose in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Parameters:

action - the action to perform.

Returns:

the new StateFuture.

thenCombine

public <U,V> InternalAsyncFuture<V> thenCombine(org.apache.flink.api.common.state.v2.StateFuture<? extends U> other,
                                                      org.apache.flink.util.function.BiFunctionWithException<? super T,? super U,? extends V,? extends Exception> fn)

Description copied from interface: InternalAsyncFuture

Returns a new StateFuture that, when this and the other given future both complete normally, is executed with the two results as arguments to the supplied function.

Specified by:

thenCombine in interface AsyncFuture<T>

Specified by:

thenCombine in interface InternalAsyncFuture<T>

Specified by:

thenCombine in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Type Parameters:

U - the type of the other StateFuture's result.

V - the function's return type.

Parameters:

other - the other StateFuture.

fn - the function to use to compute the value of the returned StateFuture.

Returns:

the new StateFuture.

thenConditionallyApply

public <U,V> InternalAsyncFuture<org.apache.flink.api.java.tuple.Tuple2<Boolean,Object>> thenConditionallyApply(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition,
                                                                                                                            org.apache.flink.util.function.FunctionWithException<? super T,? extends U,? extends Exception> actionIfTrue,
                                                                                                                            org.apache.flink.util.function.FunctionWithException<? super T,? extends V,? extends Exception> actionIfFalse)

Description copied from interface: InternalAsyncFuture

Apply a condition test on the result of this StateFuture, and try to perform one action out of two based on the result. Gather the results of the condition test and the selected action into a StateFuture of tuple. The relationship between the action result and the returned new StateFuture are just like the InternalAsyncFuture.thenApply(FunctionWithException).

Specified by:

thenConditionallyApply in interface AsyncFuture<T>

Specified by:

thenConditionallyApply in interface InternalAsyncFuture<T>

Specified by:

thenConditionallyApply in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Type Parameters:

U - the type of the output from actionIfTrue.

V - the type of the output from actionIfFalse.

Parameters:

condition - the condition test.

actionIfTrue - the function to apply if the condition returns true.

actionIfFalse - the function to apply if the condition returns false.

Returns:

the new StateFuture with the result of condition test, and result of action.

thenConditionallyApply

public <U> InternalAsyncFuture<org.apache.flink.api.java.tuple.Tuple2<Boolean,U>> thenConditionallyApply(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition,
                                                                                                               org.apache.flink.util.function.FunctionWithException<? super T,? extends U,? extends Exception> actionIfTrue)

Description copied from interface: InternalAsyncFuture

Apply a condition test on the result of this StateFuture, and try to perform the action if test result is true. Gather the results of the condition test and the action (if applied) into a StateFuture of tuple. The relationship between the action result and the returned new StateFuture are just like the InternalAsyncFuture.thenApply(FunctionWithException).

Specified by:

thenConditionallyApply in interface AsyncFuture<T>

Specified by:

thenConditionallyApply in interface InternalAsyncFuture<T>

Specified by:

thenConditionallyApply in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Type Parameters:

U - the type of the output from actionIfTrue.

Parameters:

condition - the condition test.

actionIfTrue - the function to apply if the condition returns true.

Returns:

the new StateFuture with the result of condition test, and result of action.

thenConditionallyAccept

public InternalAsyncFuture<Boolean> thenConditionallyAccept(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition,
                                                            org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> actionIfTrue,
                                                            org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> actionIfFalse)

Description copied from interface: InternalAsyncFuture

Apply a condition test on the result of this StateFuture, and try to perform one action out of two based on the result. Gather the results of the condition test StateFuture.

Specified by:

thenConditionallyAccept in interface AsyncFuture<T>

Specified by:

thenConditionallyAccept in interface InternalAsyncFuture<T>

Specified by:

thenConditionallyAccept in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Parameters:

condition - the condition test.

actionIfTrue - the function to apply if the condition returns true.

actionIfFalse - the function to apply if the condition returns false.

Returns:

the new StateFuture.

thenConditionallyAccept

public InternalAsyncFuture<Boolean> thenConditionallyAccept(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition,
                                                            org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> actionIfTrue)

Description copied from interface: InternalAsyncFuture

Apply a condition test on the result of this StateFuture, and try to perform the action if test result is true. Gather the results of the condition test StateFuture.

Specified by:

thenConditionallyAccept in interface AsyncFuture<T>

Specified by:

thenConditionallyAccept in interface InternalAsyncFuture<T>

Specified by:

thenConditionallyAccept in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Parameters:

condition - the condition test.

actionIfTrue - the function to apply if the condition returns true.

Returns:

the new StateFuture.

thenConditionallyCompose

public <U,V> InternalAsyncFuture<org.apache.flink.api.java.tuple.Tuple2<Boolean,Object>> thenConditionallyCompose(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition,
                                                                                                                              org.apache.flink.util.function.FunctionWithException<? super T,? extends org.apache.flink.api.common.state.v2.StateFuture<U>,? extends Exception> actionIfTrue,
                                                                                                                              org.apache.flink.util.function.FunctionWithException<? super T,? extends org.apache.flink.api.common.state.v2.StateFuture<V>,? extends Exception> actionIfFalse)

Description copied from interface: InternalAsyncFuture

Apply a condition test on the result of this StateFuture, and try to perform one action out of two based on the result. Gather the results of the condition test and the selected action into a StateFuture of tuple. The relationship between the action result and the returned new StateFuture are just like the InternalAsyncFuture.thenCompose(FunctionWithException).

Specified by:

thenConditionallyCompose in interface AsyncFuture<T>

Specified by:

thenConditionallyCompose in interface InternalAsyncFuture<T>

Specified by:

thenConditionallyCompose in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Type Parameters:

U - the type of the output from actionIfTrue.

V - the type of the output from actionIfFalse.

Parameters:

condition - the condition test.

actionIfTrue - the function to apply if the condition returns true.

actionIfFalse - the function to apply if the condition returns false.

Returns:

the new StateFuture with the result of condition test, and result of action.

thenConditionallyCompose

public <U> InternalAsyncFuture<org.apache.flink.api.java.tuple.Tuple2<Boolean,U>> thenConditionallyCompose(org.apache.flink.util.function.FunctionWithException<? super T,Boolean,? extends Exception> condition,
                                                                                                                 org.apache.flink.util.function.FunctionWithException<? super T,? extends org.apache.flink.api.common.state.v2.StateFuture<U>,? extends Exception> actionIfTrue)

Description copied from interface: InternalAsyncFuture

Apply a condition test on the result of this StateFuture, and try to perform the action if test result is true. Gather the results of the condition test and the action (if applied) into a StateFuture of tuple. The relationship between the action result and the returned new StateFuture are just like the InternalAsyncFuture.thenCompose(FunctionWithException) (FunctionWithException)}.

Specified by:

thenConditionallyCompose in interface AsyncFuture<T>

Specified by:

thenConditionallyCompose in interface InternalAsyncFuture<T>

Specified by:

thenConditionallyCompose in interface org.apache.flink.api.common.state.v2.StateFuture<T>

Type Parameters:

U - the type of the output from actionIfTrue.

Parameters:

condition - the condition test.

actionIfTrue - the function to apply if the condition returns true.

Returns:

the new StateFuture with the result of condition test, and result of action.

makeNewFuture

public <A> AsyncFutureImpl<A> makeNewFuture()

Make a new future based on context of this future. Subclasses need to overload this method to generate their own instances (if needed).

Returns:

the new created future.

isDone

public boolean isDone()

Description copied from interface: InternalAsyncFuture

Returns true if completed in any fashion: normally, exceptionally, or via cancellation.

Specified by:

isDone in interface InternalAsyncFuture<T>

Returns:

true if completed

get

public T get()

Description copied from interface: InternalAsyncFuture

Waits if necessary for the computation to complete, and then retrieves its result.

Specified by:

get in interface InternalAsyncFuture<T>

complete

public void complete(T result)

Description copied from interface: InternalAsyncFuture

Complete this future.

Specified by:

complete in interface InternalAsyncFuture<T>

completeExceptionally

public void completeExceptionally(String message,
                                  Throwable ex)

Description copied from interface: InternalAsyncFuture

Fail this future and pass the given exception to the runtime.

Specified by:

completeExceptionally in interface InternalAsyncFuture<T>

Parameters:

message - the description of this exception

ex - the exception

callbackRegistered

public void callbackRegistered()

Will be triggered when a callback is registered.

postComplete

public void postComplete(boolean inCallbackRunner)

Will be triggered when this future completes.

callbackFinished

public void callbackFinished()

Will be triggered when a callback finishes processing.

thenSyncAccept

public void thenSyncAccept(org.apache.flink.util.function.ThrowingConsumer<? super T,? extends Exception> action)

Description copied from interface: InternalAsyncFuture

Accept the action in the same thread with the one of complete (or current thread if it has been completed).

Specified by:

thenSyncAccept in interface InternalAsyncFuture<T>

Parameters:

action - the action to perform.