@PublicEvolving
public interface StateFuture<T> {
    /**
     * Returns a new StateFuture that, when this future completes
     * normally, is executed with this future's result as the argument
     * to the supplied function.
     *
     * @param fn the function to use to compute the value of
     * the returned StateFuture
     * @param <U> the function's return type
     * @return the new StateFuture
     */
    <U> StateFuture<U> thenApply(Function<? super T,? extends U> fn);

    /**
     * Returns a new StateFuture that, when this future completes
     * normally, is executed with this future's result as the argument
     * to the supplied action.
     *
     * @param action the action to perform before completing the
     * returned StateFuture
     * @return the new StateFuture
     */
    StateFuture<Void> thenAccept(Consumer<? super T> action);

    /**
     * Returns a new future that, when this future completes normally,
     * is executed with this future as the argument to the supplied function.
     *
     * @param action the action to perform 
     * @return the new StateFuture
     */
    <U> StateFuture<U> thenCompose(
        Function<? super T, ? extends StateFuture<U>> action);


    /**
     * 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.
     *
     * @param other the other StateFuture
     * @param fn the function to use to compute the value of
     * the returned StateFuture
     * @param <U> the type of the other StateFuture's result
     * @param <V> the function's return type
     * @return the new StateFuture
     */
    <U,V> StateFuture<V> thenCombine
    (StateFuture<? extends U> other,
     BiFunction<? super T,? super U,? extends V> fn);
}

@PublicEvolving
public class StateFutureUtils {
    /**
     * Returns a completed future that does nothing and return null.
     */
    public static <V> StateFuture<V> completedVoidFuture() {
        // xxx
    }

    /** Returns a completed future that does nothing and return provided result. */
    public static <V> StateFuture<V> completedFuture(V result) {
        // xxx
    }

     /**
     * Creates a future that is complete once multiple other futures completed. Upon successful
     * completion, the future returns the collection of the futures' results.
     *
     * @param futures The futures that make up the conjunction. No null entries are allowed,
     *        otherwise a IllegalArgumentException will be thrown.
     * @return The StateFuture that completes once all given futures are complete.
     */
    public static <T> StateFuture<Collection<T>> combineAll(
            Collection<? extends StateFuture<? extends T>> futures) {
        // xxx
    }
}

@PublicEvolving
public interface StateIterator<T> {
    /**
     * Async iterate the data and call the callback when data is ready.
     * @param iterating the data action when it is ready. The return is the state future for
     * chaining.
     * @param <U> the type of the inner returned StateFuture's result
     * @return the Future that will trigger when this iterator and all returned state future get its
     * results.
     */
    <U> StateFuture<Collection<U>> onNext(Function<T, StateFuture<? extends U>> iterating);


    /**
     * Async iterate the data and call the callback when data is ready.
     * @param iterating the data action when it is ready
     * @return the Future that will trigger when this iterator ends.
     */
    StateFuture<Void> onNext(Consumer<T> iterating);

    /** Return if this iterator is empty synchronously. */
    boolean isEmpty();
}

@PublicEvolving
public interface State {

    /** Removes the value mapped under the current key. */
    StateFuture<Void> asyncClear();
}

@PublicEvolving
public interface ValueState<T> extends State {
    /**
     * Returns the current value for the state asynchronously. When the state is not partitioned the
     * returned value is the same for all inputs in a given operator instance. If state partitioning
     * is applied, the value returned depends on the current operator input, as the operator maintains
     * an independent state for each partition. When no value was previously set using {@link #asyncUpdate(Object)},
     * this will return {@code null} asynchronously.
     *
     * @return The {@link StateFuture} that will return the value corresponding to the current input.
     */
     StateFuture<T> asyncValue();

    /**
     * Updates the operator state accessible by {@link #asyncValue()} to the given value. The next time
     * {@link #asyncValue()} is called (for the same state partition) the returned state will represent
     * the updated value. When a partitioned state is updated with {@code null}, the state for the
     * current key will be removed.
     *
     * @param value The new value for the state.
     * @return The {@link StateFuture} that will trigger the callback when update finishes. 
     */
     StateFuture<Void> asyncUpdate(T value);
}

@PublicEvolving
public interface ListState<T> extends State {

    /**
     * Returns the current iterator for the state asynchronously. When the state is not partitioned the
     * returned value is the same for all inputs in a given operator instance. If state partitioning
     * is applied, the value returned depends on the current operator input, as the operator maintains
     * an independent state for each partition. This method will never return (@code null).
     *
     * @return The {@link StateFuture} that will return list iterator corresponding to the current input. 
    */
    StateFuture<StateIterator<T>> asyncGet();

    /**
     * Updates the operator state accessible by {@link #asyncGet()} by adding the given value to the list
     * of values. The next time {@link #asyncGet()} is called (for the same state partition) the returned
     * state will represent the updated list.
     *
     * <p> Null is not allowed to be passed in
     *
     * @param value The new value for the state.
     * @return The {@link StateFuture} that will trigger the callback when update finishes.  
     */
    StateFuture<Void> asyncAdd(T value);

    /**
     * Updates the operator state accessible by {@link #asyncGet()} by updating existing values to the
     * given list of values. The next time {@link #asyncGet()} is called (for the same state partition)
     * the returned state will represent the updated list.
     *
     * <p> Null value passed in or any null value in list is not allowed.
     *
     * @param values The new value list for the state.
     * @return The {@link StateFuture} that will trigger the callback when update finishes.  
     */
    StateFuture<Void> asyncUpdate(List<T> values);

    /**
     * Updates the operator state accessible by {@link #asyncGet()} by adding the given values to the list
     * of values. The next time {@link #asyncGet()} is called (for the same state partition) the returned
     * state will represent the updated list.
     *
     * <p> Null value passed in or any null value in list is not allowed.
     *
     * @param values The new values to add for the state.
     * @return The {@link StateFuture} that will trigger the callback when update finishes.  
     */
    StateFuture<Void> asyncAddAll(List<T> values);
}

@PublicEvolving
public interface MapState<UK, UV> extends State {

    /**
     * Returns the current value associated with the given key asynchronously. When the state is 
     * not partitioned the returned value is the same for all inputs in a given operator instance.
     * If state partitioning is applied, the value returned depends on the current operator input,
     * as the operator maintains an independent state for each partition.
     *
     * @return The {@link StateFuture} that will return value corresponding to the current input. 
     */
    StateFuture<UV> asyncGet(UK key);

    /**
     * Update the current value associated with the given key asynchronously. When the state is 
     * not partitioned the value is updated for all inputs in a given operator instance.
     * If state partitioning is applied, the updated value depends on the current operator input,
     * as the operator maintains an independent state for each partition.
     *
     * @param key The key that will be updated.
     * @param value The new value for the key.  
     * @return The {@link StateFuture} that will trigger the callback when update finishes.  
     */
    StateFuture<Void> asyncPut(UK key, UV value);

    /**
     * Update all of the mappings from the given map into the state asynchronously. When the state is 
     * not partitioned the value is updated for all inputs in a given operator instance.
     * If state partitioning is applied, the updated mapping depends on the current operator input,
     * as the operator maintains an independent state for each partition.
     *
     * @param map The mappings to be stored in this state.
     * @return The {@link StateFuture} that will trigger the callback when update finishes.  
     */ 
    StateFuture<Void> asyncPutAll(Map<UK, UV> map);

    /**
     * Delete the mapping of the given key from the state asynchronously. When the state is 
     * not partitioned the deleted value is the same for all inputs in a given operator instance.
     * If state partitioning is applied, the value deleted depends on the current operator input,
     * as the operator maintains an independent state for each partition.
     *
     * @param key The key of the mapping.
     * @return The {@link StateFuture} that will trigger the callback when update finishes.  
     */
    StateFuture<Void> asyncRemove(UK key);

    /**
     * Returns whether there exists the given mapping asynchronously. When the state is 
     * not partitioned the returned value is the same for all inputs in a given operator instance.
     * If state partitioning is applied, the value returned depends on the current operator input,
     * as the operator maintains an independent state for each partition.
     *
     * @param key The key of the mapping.
     * @return The {@link StateFuture} that will return true if there exists a mapping whose key
     * equals to the given key
     */
    StateFuture<Boolean> asyncContains(UK key);

    /**
     * Returns the current iterator for all the mappings of this state asynchronously. When the state
     * is not partitioned the returned iterator is the same for all inputs in a given operator instance.
     * If state partitioning is applied, the iterator returned depends on the current operator input,
     * as the operator maintains an independent state for each partition.
     *
     * @return The {@link StateFuture} that will return mapping iterator corresponding to the current input. 
    */
    StateFuture<StateIterator<Map.Entry<UK, UV>>> asyncEntries();

    /**
     * Returns the current iterator for all the keys of this state asynchronously. When the state
     * is not partitioned the returned iterator is the same for all inputs in a given operator instance.
     * If state partitioning is applied, the iterator returned depends on the current operator input,
     * as the operator maintains an independent state for each partition.
     *
     * @return The {@link StateFuture} that will return key iterator corresponding to the current input. 
    */
    StateFuture<StateIterator<UK>> asyncKeys();

    /**
     * Returns the current iterator for all the values of this state asynchronously. When the state
     * is not partitioned the returned iterator is the same for all inputs in a given operator instance.
     * If state partitioning is applied, the iterator returned depends on the current operator input,
     * as the operator maintains an independent state for each partition.
     *
     * @return The {@link StateFuture} that will return value iterator corresponding to the current input. 
    */
    StateFuture<StateIterator<UV>> asyncValues();

    /**
     * Returns whether this state contains no key-value mappings asynchronously. When the state is 
     * not partitioned the returned value is the same for all inputs in a given operator instance.
     * If state partitioning is applied, the value returned depends on the current operator input,
     * as the operator maintains an independent state for each partition. 
     *
     * @return The {@link StateFuture} that will return true if there is no key-value mapping, otherwise false.
     */
    StateFuture<Boolean> asyncIsEmpty();
}

private final ValueState<RowData> aState;

private final MapState<RowData, Tuple2<RowData, Integer>> bState;

private final KeySelector<RowData, RowData> uniqueKeySelector;

public void processElement1(StreamRecord<RowData> element) throws Exception {
    RowData input = element.getValue();
    boolean isAccumulateMsg = RowDataUtil.isAccumulateMsg(input);
    if (isAccumulateMsg) {
        bState.asyncValues().thenCompose((iterator) -> {
            return iterator.onHasNext((value) -> {
                // value is Tuple2<RowData, Integer>
                if (joinCondition.apply(input, value.f0)) {
                    output(input, value.f0);
                    RowData key = uniqueKeySelector.getKey(value.f0);
                    return bState.asyncPut(key, Tuple2.of(value.f0, value.f1 + 1));
                } else {
                    return FutureUtils.emptyFuture();
                }
            });
        }).thenAccept((e) -> {
            return aState.asyncUpdate(input);
        });
        // or just:
        // aState.asyncUpdate(input);
    } else {
        // omit retraction
    }
}

bState.asyncValues().thenApply((iterator) -> {
    iterator.onHasNext((value) -> {
         if (joinCondition.apply(input, value.f0)) {
             output(input, value.f0);
             RowData key = uniqueKeySelector.getKey(value.f0);
             bState.put(key, Tuple2.of(value.f0, value.f1 + 1));
         }
    });
});
aState.update(input);