flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From trohrm...@apache.org
Subject [49/50] [abbrv] flink git commit: [FLINK-4361] Introduce Flink's own future abstraction
Date Wed, 21 Sep 2016 09:53:22 GMT
[FLINK-4361] Introduce Flink's own future abstraction

Flink's future abstraction whose API is similar to Java 8's CompletableFuture.
That's in order to ease a future transition to this class once we ditch Java 7.
The current set of operations comprises:

- isDone to check the completion of the future
- get/getNow to obtain the future's value
- cancel to cancel the future (best effort basis)
- thenApplyAsync to transform the future's value into another value
- thenAcceptAsync to register a callback for a successful completion of the future
- exceptionallyAsync to register a callback for an exception completion of the future
- thenComposeAsync to transform the future's value and flatten the returned future
- handleAsync to register a callback which is called either with the regular result
or the exceptional result

Additionally, Flink offers a CompletableFuture which can be completed with a regular
value or an exception:

- complete/completeExceptionally

Complete FlinkCompletableFuture exceptionally with a CanellationException upon cancel

This closes #2472.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f766f120
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f766f120
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f766f120

Branch: refs/heads/flip-6
Commit: f766f120e4af6b60a3e628836963eb981005325c
Parents: 0a134a3
Author: Till Rohrmann <trohrmann@apache.org>
Authored: Fri Sep 2 21:13:34 2016 +0200
Committer: Till Rohrmann <trohrmann@apache.org>
Committed: Wed Sep 21 11:39:18 2016 +0200

----------------------------------------------------------------------
 .../runtime/concurrent/AcceptFunction.java      |  34 +++
 .../flink/runtime/concurrent/ApplyFunction.java |  36 +++
 .../flink/runtime/concurrent/BiFunction.java    |  38 +++
 .../runtime/concurrent/CompletableFuture.java   |  47 ++++
 .../apache/flink/runtime/concurrent/Future.java | 156 +++++++++++
 .../concurrent/impl/FlinkCompletableFuture.java |  71 +++++
 .../runtime/concurrent/impl/FlinkFuture.java    | 273 +++++++++++++++++++
 .../runtime/concurrent/FlinkFutureTest.java     | 269 ++++++++++++++++++
 8 files changed, 924 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/f766f120/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/AcceptFunction.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/AcceptFunction.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/AcceptFunction.java
new file mode 100644
index 0000000..a300647
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/AcceptFunction.java
@@ -0,0 +1,34 @@
+/*
+ * 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.runtime.concurrent;
+
+/**
+ * Function which is called with a single argument and does not return a value.
+ *
+ * @param <T> type of the argument
+ */
+public interface AcceptFunction<T> {
+
+	/**
+	 * Method which handles the function call.
+	 *
+	 * @param value is the function's argument
+	 */
+	void accept(T value);
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f766f120/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/ApplyFunction.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/ApplyFunction.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/ApplyFunction.java
new file mode 100644
index 0000000..64def98
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/ApplyFunction.java
@@ -0,0 +1,36 @@
+/*
+ * 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.runtime.concurrent;
+
+/**
+ * Function which is called with a single argument.
+ *
+ * @param <V> type of the argument
+ * @param <R> type of the return value
+ */
+public interface ApplyFunction<V, R> {
+
+	/**
+	 * Method which handles the function call.
+	 *
+	 * @param value is the single argument
+	 * @return the function value
+	 */
+	R apply(V value);
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f766f120/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/BiFunction.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/BiFunction.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/BiFunction.java
new file mode 100644
index 0000000..2b09de8
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/BiFunction.java
@@ -0,0 +1,38 @@
+/*
+ * 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.runtime.concurrent;
+
+/**
+ * Function which is called with two arguments and returns a value.
+ *
+ * @param <T> type of the first argument
+ * @param <U> type of the second argument
+ * @param <R> type of the return value
+ */
+public interface BiFunction<T, U, R> {
+
+	/**
+	 * Method which handles the function call.
+	 *
+	 * @param t first argument
+	 * @param u second argument
+	 * @return the function value
+	 */
+	R apply(T t, U u);
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f766f120/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/CompletableFuture.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/CompletableFuture.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/CompletableFuture.java
new file mode 100644
index 0000000..5288bf2
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/CompletableFuture.java
@@ -0,0 +1,47 @@
+/*
+ * 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.runtime.concurrent;
+
+/**
+ * Flink's completable future abstraction. A completable future can be completed with a regular
+ * value or an exception.
+ *
+ * @param <T> type of the future's value
+ */
+public interface CompletableFuture<T> extends Future<T> {
+
+	/**
+	 * Completes the future with the given value. The complete operation only succeeds if the
future
+	 * has not been completed before. Whether it is successful or not is returned by the method.
+	 *
+	 * @param value to complete the future with
+	 * @return true if the completion was successful; otherwise false
+	 */
+	boolean complete(T value);
+
+	/**
+	 * Completes the future with the given exception. The complete operation only succeeds if
the
+	 * future has not been completed before. Whether it is successful or not is returned by
the
+	 * method.
+	 *
+	 * @param t the exception to complete the future with
+	 * @return true if the completion was successful; otherwise false
+	 */
+	boolean completeExceptionally(Throwable t);
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f766f120/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Future.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Future.java b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Future.java
new file mode 100644
index 0000000..b32bcd4
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Future.java
@@ -0,0 +1,156 @@
+/*
+ * 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.runtime.concurrent;
+
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Flink's basic future abstraction. A future represents an asynchronous operation whose
result
+ * will be contained in this instance upon completion.
+ *
+ * @param <T> type of the future's result
+ */
+public interface Future<T> {
+
+	/**
+	 * Checks if the future has been completed. A future is completed, if the result has been
+	 * delivered.
+	 *
+	 * @return true if the future is completed; otherwise false
+	 */
+	boolean isDone();
+
+	/**
+	 * Tries to cancel the future's operation. Note that not all future operations can be canceled.
+	 * The result of the cancelling will be returned.
+	 *
+	 * @param mayInterruptIfRunning true iff the future operation may be interrupted
+	 * @return true if the cancelling was successful; otherwise false
+	 */
+	boolean cancel(boolean mayInterruptIfRunning);
+
+	/**
+	 * Gets the result value of the future. If the future has not been completed, then this
+	 * operation will block indefinitely until the result has been delivered.
+	 *
+	 * @return the result value
+	 * @throws CancellationException if the future has been cancelled
+	 * @throws InterruptedException if the current thread was interrupted while waiting for
the result
+	 * @throws ExecutionException if the future has been completed with an exception
+	 */
+	T get() throws InterruptedException, ExecutionException;
+
+	/**
+	 * Gets the result value of the future. If the future has not been done, then this operation
+	 * will block the given timeout value. If the result has not been delivered within the timeout,
+	 * then the method throws an {@link TimeoutException}.
+	 *
+	 * @param timeout the time to wait for the future to be done
+	 * @param unit time unit for the timeout argument
+	 * @return the result value
+	 * @throws CancellationException if the future has been cancelled
+	 * @throws InterruptedException if the current thread was interrupted while waiting for
the result
+	 * @throws ExecutionException if the future has been completed with an exception
+	 * @throws TimeoutException if the future has not been completed within the given timeout
+	 */
+	T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
+
+	/**
+	 * Gets the value of the future. If the future has not been completed when calling this
+	 * function, the given value is returned.
+	 *
+	 * @param valueIfAbsent value which is returned if the future has not been completed
+	 * @return value of the future or the given value if the future has not been completed
+	 * @throws ExecutionException if the future has been completed with an exception
+	 */
+	T getNow(T valueIfAbsent) throws ExecutionException;
+
+	/**
+	 * Applies the given function to the value of the future. The result of the apply function
is
+	 * the value of the newly returned future.
+	 * <p>
+	 * The apply function is executed asynchronously by the given executor.
+	 *
+	 * @param applyFunction function to apply to the future's value
+	 * @param executor used to execute the given apply function asynchronously
+	 * @param <R> type of the apply function's return value
+	 * @return future representing the return value of the given apply function
+	 */
+	<R> Future<R> thenApplyAsync(ApplyFunction<? super T, ? extends R> applyFunction,
Executor executor);
+
+	/**
+	 * Applies the accept function to the value of the future. Unlike the {@link ApplyFunction},
the
+	 * {@link AcceptFunction} does not return a value. The returned future, thus, represents
only
+	 * the completion of the accept callback.
+	 * <p>
+	 * The accept function is executed asynchronously by the given executor.
+	 *
+	 * @param acceptFunction function to apply to the future's value
+	 * @param executor used to execute the given apply function asynchronously
+	 * @return future representing the completion of the accept callback
+	 */
+	Future<Void> thenAcceptAsync(AcceptFunction<? super T> acceptFunction, Executor
executor);
+
+	/**
+	 * Applies the given function to the value of the future if the future has been completed
+	 * exceptionally. The completing exception is given to the apply function which can return
a new
+	 * value which is the value of the returned future.
+	 * <p>
+	 * The apply function is executed asynchronously by the given executor.
+	 *
+	 * @param exceptionallyFunction to apply to the future's value if it is an exception
+	 * @param executor used to execute the given apply function asynchronously
+	 * @param <R> type of the apply function's return value
+	 * @return future representing the return value of the given apply function
+	 */
+	<R> Future<R> exceptionallyAsync(ApplyFunction<Throwable, ? extends R>
exceptionallyFunction, Executor executor);
+
+	/**
+	 * Applies the given function to the value of the future. The apply function returns a future
+	 * result, which is flattened. This means that the resulting future of this method represents
+	 * the future's value of the apply function.
+	 * <p>
+	 * The apply function is executed asynchronously by the given executor.
+	 *
+	 * @param composeFunction to apply to the future's value. The function returns a future
which is
+	 *                        flattened
+	 * @param executor used to execute the given apply function asynchronously
+	 * @param <R> type of the returned future's value
+	 * @return future representing the flattened return value of the apply function
+	 */
+	<R> Future<R> thenComposeAsync(ApplyFunction<? super T, Future<? extends
R>> composeFunction, Executor executor);
+
+	/**
+	 * Applies the given handle function to the result of the future. The result can either
be the
+	 * future's value or the exception with which the future has been completed. The two cases
are
+	 * mutually exclusive. The result of the handle function is the returned future's value.
+	 * <p>
+	 * The handle function is executed asynchronously by the given executor.
+	 *
+	 * @param biFunction applied to the result (normal and exceptional) of the future
+	 * @param executor used to execute the handle function asynchronously
+	 * @param <R> type of the handle function's return value
+	 * @return future representing the handle function's return value
+	 */
+	<R> Future<R> handleAsync(BiFunction<? super T, Throwable, ? extends R>
biFunction, Executor executor);
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f766f120/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/impl/FlinkCompletableFuture.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/impl/FlinkCompletableFuture.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/impl/FlinkCompletableFuture.java
new file mode 100644
index 0000000..5566880
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/impl/FlinkCompletableFuture.java
@@ -0,0 +1,71 @@
+/*
+ * 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.runtime.concurrent.impl;
+
+import org.apache.flink.runtime.concurrent.CompletableFuture;
+import org.apache.flink.util.Preconditions;
+import scala.concurrent.Promise;
+
+import java.util.concurrent.CancellationException;
+
+/**
+ * Implementation of {@link CompletableFuture} which is backed by {@link Promise}.
+ *
+ * @param <T> type of the future's value
+ */
+public class FlinkCompletableFuture<T> extends FlinkFuture<T> implements CompletableFuture<T>
{
+
+	private final Promise<T> promise;
+
+	public FlinkCompletableFuture() {
+		promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
+		scalaFuture = promise.future();
+	}
+
+	@Override
+	public boolean complete(T value) {
+		Preconditions.checkNotNull(value);
+
+		try {
+			promise.success(value);
+
+			return true;
+		} catch (IllegalStateException e) {
+			return false;
+		}
+	}
+
+	@Override
+	public boolean completeExceptionally(Throwable t) {
+		Preconditions.checkNotNull(t);
+
+		try {
+			promise.failure(t);
+
+			return true;
+		} catch (IllegalStateException e) {
+			return false;
+		}
+	}
+
+	@Override
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		return completeExceptionally(new CancellationException("Future has been canceled."));
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f766f120/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/impl/FlinkFuture.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/impl/FlinkFuture.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/impl/FlinkFuture.java
new file mode 100644
index 0000000..361cd3d
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/impl/FlinkFuture.java
@@ -0,0 +1,273 @@
+/*
+ * 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.runtime.concurrent.impl;
+
+import akka.dispatch.ExecutionContexts$;
+import akka.dispatch.Futures;
+import akka.dispatch.Mapper;
+import akka.dispatch.Recover;
+import org.apache.flink.runtime.concurrent.AcceptFunction;
+import org.apache.flink.runtime.concurrent.ApplyFunction;
+import org.apache.flink.runtime.concurrent.Future;
+import org.apache.flink.runtime.concurrent.BiFunction;
+import org.apache.flink.util.Preconditions;
+import scala.Option;
+import scala.concurrent.Await;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.duration.Duration;
+import scala.concurrent.duration.FiniteDuration;
+import scala.util.Failure;
+import scala.util.Success;
+import scala.util.Try;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Implementation of {@link Future} which is backed by {@link scala.concurrent.Future}.
+ *
+ * @param <T> type of the future's value
+ */
+public class FlinkFuture<T> implements Future<T> {
+
+	protected scala.concurrent.Future<T> scalaFuture;
+
+	FlinkFuture() {
+		scalaFuture = null;
+	}
+
+	public FlinkFuture(scala.concurrent.Future<T> scalaFuture) {
+		this.scalaFuture = Preconditions.checkNotNull(scalaFuture);
+	}
+
+	//-----------------------------------------------------------------------------------
+	// Future's methods
+	//-----------------------------------------------------------------------------------
+
+	@Override
+	public boolean isDone() {
+		return scalaFuture.isCompleted();
+	}
+
+	@Override
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		return false;
+	}
+
+	@Override
+	public T get() throws InterruptedException, ExecutionException {
+		Preconditions.checkNotNull(scalaFuture);
+
+		try {
+			return Await.result(scalaFuture, Duration.Inf());
+		} catch (InterruptedException e) {
+			throw e;
+		} catch (Exception e) {
+			throw new ExecutionException(e);
+		}
+	}
+
+	@Override
+	public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException,
TimeoutException {
+		Preconditions.checkNotNull(scalaFuture);
+		Preconditions.checkArgument(timeout >= 0L, "The timeout value has to be larger or "
+
+			"equal than 0.");
+
+		try {
+			return Await.result(scalaFuture, new FiniteDuration(timeout, unit));
+		} catch (InterruptedException | TimeoutException e) {
+			throw e;
+		} catch (Exception e) {
+			throw new ExecutionException(e);
+		}
+	}
+
+	@Override
+	public T getNow(T valueIfAbsent) throws ExecutionException {
+		Preconditions.checkNotNull(scalaFuture);
+		Preconditions.checkNotNull(valueIfAbsent);
+
+		Option<Try<T>> value = scalaFuture.value();
+
+		if (value.isDefined()) {
+			Try<T> tri = value.get();
+
+			if (tri instanceof Success) {
+				return ((Success<T>)tri).value();
+			} else {
+				throw new ExecutionException(((Failure<T>)tri).exception());
+			}
+		} else {
+			return valueIfAbsent;
+		}
+	}
+
+	@Override
+	public <R> Future<R> thenApplyAsync(final ApplyFunction<? super T, ? extends
R> applyFunction, Executor executor) {
+		Preconditions.checkNotNull(scalaFuture);
+		Preconditions.checkNotNull(applyFunction);
+		Preconditions.checkNotNull(executor);
+
+		scala.concurrent.Future<R> mappedFuture = scalaFuture.map(new Mapper<T, R>()
{
+			@Override
+			public R apply(T value) {
+				return applyFunction.apply(value);
+			}
+		}, createExecutionContext(executor));
+
+		return new FlinkFuture<>(mappedFuture);
+	}
+
+	@Override
+	public Future<Void> thenAcceptAsync(final AcceptFunction<? super T> acceptFunction,
Executor executor) {
+		Preconditions.checkNotNull(scalaFuture);
+		Preconditions.checkNotNull(acceptFunction);
+		Preconditions.checkNotNull(executor);
+
+		scala.concurrent.Future<Void> acceptedFuture = scalaFuture.map(new Mapper<T, Void>()
{
+			@Override
+			public Void apply(T value) {
+				acceptFunction.accept(value);
+
+				return null;
+			}
+		}, createExecutionContext(executor));
+
+		return new FlinkFuture<>(acceptedFuture);
+	}
+
+	@Override
+	public <R> Future<R> exceptionallyAsync(final ApplyFunction<Throwable, ?
extends R> exceptionallyFunction, Executor executor) {
+		Preconditions.checkNotNull(scalaFuture);
+		Preconditions.checkNotNull(exceptionallyFunction);
+		Preconditions.checkNotNull(executor);
+
+		scala.concurrent.Future<R> recoveredFuture = scalaFuture.recover(new Recover<R>()
{
+			@Override
+			public R recover(Throwable failure) throws Throwable {
+				return exceptionallyFunction.apply(failure);
+			}
+		}, createExecutionContext(executor));
+
+		return new FlinkFuture<>(recoveredFuture);
+	}
+
+	@Override
+	public <R> Future<R> thenComposeAsync(final ApplyFunction<? super T, Future<?
extends R>> applyFunction, final Executor executor) {
+		Preconditions.checkNotNull(scalaFuture);
+		Preconditions.checkNotNull(applyFunction);
+		Preconditions.checkNotNull(executor);
+
+		scala.concurrent.Future<R> flatMappedFuture = scalaFuture.flatMap(new Mapper<T,
scala.concurrent.Future<R>>() {
+			@Override
+			public scala.concurrent.Future<R> apply(T value) {
+				final Future<? extends R> future = applyFunction.apply(value);
+
+				if (future instanceof FlinkFuture) {
+					@SuppressWarnings("unchecked")
+					FlinkFuture<R> flinkFuture = (FlinkFuture<R>) future;
+
+					return flinkFuture.scalaFuture;
+				} else {
+					return Futures.future(new Callable<R>() {
+						@Override
+						public R call() throws Exception {
+							return future.get();
+						}
+					}, createExecutionContext(executor));
+				}
+			}
+		}, createExecutionContext(executor));
+
+		return new FlinkFuture<>(flatMappedFuture);
+	}
+
+	@Override
+	public <R> Future<R> handleAsync(final BiFunction<? super T, Throwable, ?
extends R> biFunction, Executor executor) {
+		Preconditions.checkNotNull(scalaFuture);
+		Preconditions.checkNotNull(biFunction);
+		Preconditions.checkNotNull(executor);
+
+		scala.concurrent.Future<R> mappedFuture = scalaFuture.map(new Mapper<T, R>()
{
+			@Override
+			public R checkedApply(T value) throws Exception {
+				try {
+					return biFunction.apply(value, null);
+				} catch (Throwable t) {
+					throw new FlinkFuture.WrapperException(t);
+				}
+			}
+		}, createExecutionContext(executor));
+
+		scala.concurrent.Future<R> recoveredFuture = mappedFuture.recover(new Recover<R>()
{
+			@Override
+			public R recover(Throwable failure) throws Throwable {
+				if (failure instanceof FlinkFuture.WrapperException) {
+					throw failure.getCause();
+				} else {
+					return biFunction.apply(null, failure);
+				}
+			}
+		}, createExecutionContext(executor));
+
+
+		return new FlinkFuture<>(recoveredFuture);
+	}
+
+	//-----------------------------------------------------------------------------------
+	// Static factory methods
+	//-----------------------------------------------------------------------------------
+
+	/**
+	 * Creates a future whose value is determined by the asynchronously executed callable.
+	 *
+	 * @param callable whose value is delivered by the future
+	 * @param executor to be used to execute the callable
+	 * @param <T> type of the future's value
+	 * @return future which represents the value of the callable
+	 */
+	public static <T> Future<T> supplyAsync(Callable<T> callable, Executor
executor) {
+		Preconditions.checkNotNull(callable);
+		Preconditions.checkNotNull(executor);
+
+		scala.concurrent.Future<T> scalaFuture = Futures.future(callable, createExecutionContext(executor));
+
+		return new FlinkFuture<>(scalaFuture);
+	}
+
+	//-----------------------------------------------------------------------------------
+	// Helper functions and types
+	//-----------------------------------------------------------------------------------
+
+	private static ExecutionContext createExecutionContext(Executor executor) {
+		return ExecutionContexts$.MODULE$.fromExecutor(executor);
+	}
+
+	private static class WrapperException extends Exception {
+
+		private static final long serialVersionUID = 6533166370660884091L;
+
+		WrapperException(Throwable cause) {
+			super(cause);
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/f766f120/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FlinkFutureTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FlinkFutureTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FlinkFutureTest.java
new file mode 100644
index 0000000..bd5af66
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FlinkFutureTest.java
@@ -0,0 +1,269 @@
+/*
+ * 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.runtime.concurrent;
+
+import org.apache.flink.runtime.concurrent.impl.FlinkCompletableFuture;
+import org.apache.flink.runtime.concurrent.impl.FlinkFuture;
+import org.apache.flink.util.TestLogger;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for Flink's future implementation.
+ */
+public class FlinkFutureTest extends TestLogger {
+
+	private static ExecutorService executor;
+
+	@BeforeClass
+	public static void setup() {
+		executor = Executors.newSingleThreadExecutor();
+	}
+
+	@AfterClass
+	public static void teardown() {
+		executor.shutdown();
+	}
+
+	@Test
+	public void testFutureApply() throws Exception {
+		int expectedValue = 42;
+
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+
+		Future<String> appliedFuture = initialFuture.thenApplyAsync(new ApplyFunction<Integer,
String>() {
+			@Override
+			public String apply(Integer value) {
+				return String.valueOf(value);
+			}
+		}, executor);
+
+		initialFuture.complete(expectedValue);
+
+		assertEquals(String.valueOf(expectedValue), appliedFuture.get());
+	}
+
+	@Test(expected = TimeoutException.class)
+	public void testFutureGetTimeout() throws InterruptedException, ExecutionException, TimeoutException
{
+		CompletableFuture<Integer> future = new FlinkCompletableFuture<>();
+
+		future.get(10, TimeUnit.MILLISECONDS);
+
+		fail("Get should have thrown a timeout exception.");
+	}
+
+	@Test(expected = TestException.class)
+	public void testExceptionalCompletion() throws Throwable {
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+
+		initialFuture.completeExceptionally(new TestException("Test exception"));
+
+		try {
+			initialFuture.get();
+
+			fail("Get should have thrown an exception.");
+		} catch (ExecutionException e) {
+			throw e.getCause();
+		}
+	}
+
+	/**
+	 * Tests that an exception is propagated through an apply function.
+	 */
+	@Test(expected = TestException.class)
+	public void testExceptionPropagation() throws Throwable {
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+
+		Future<String> mappedFuture = initialFuture.thenApplyAsync(new ApplyFunction<Integer,
String>() {
+			@Override
+			public String apply(Integer value) {
+				throw new TestException("Test exception");
+			}
+		}, executor);
+
+		Future<String> mapped2Future = mappedFuture.thenApplyAsync(new ApplyFunction<String,
String>() {
+			@Override
+			public String apply(String value) {
+				return "foobar";
+			}
+		}, executor);
+
+		initialFuture.complete(42);
+
+		try {
+			mapped2Future.get();
+
+			fail("Get should have thrown an exception.");
+		} catch (ExecutionException e) {
+			throw e.getCause();
+		}
+	}
+
+	@Test
+	public void testExceptionally() throws ExecutionException, InterruptedException {
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+		String exceptionMessage = "Foobar";
+
+		Future<String> recovered = initialFuture.exceptionallyAsync(new ApplyFunction<Throwable,
String>() {
+			@Override
+			public String apply(Throwable value) {
+				return value.getMessage();
+			}
+		}, executor);
+
+		initialFuture.completeExceptionally(new TestException(exceptionMessage));
+
+		String actualMessage = recovered.get();
+
+		assertEquals(exceptionMessage, actualMessage);
+	}
+
+	@Test
+	public void testCompose() throws ExecutionException, InterruptedException {
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+
+		final int expectedValue = 42;
+
+		Future<Integer> composedFuture = initialFuture.thenComposeAsync(new ApplyFunction<Integer,
Future<? extends Integer>>() {
+			@Override
+			public Future<? extends Integer> apply(Integer value) {
+				return FlinkFuture.supplyAsync(new Callable<Integer>() {
+					@Override
+					public Integer call() throws Exception {
+						return expectedValue;
+					}
+				}, executor);
+			}
+		}, executor);
+
+		initialFuture.complete(42);
+
+		int actualValue = composedFuture.get();
+
+		assertEquals(expectedValue, actualValue);
+	}
+
+	@Test
+	public void testGetNow() throws ExecutionException {
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+
+		final int absentValue = 41;
+
+		assertEquals(new Integer(absentValue), initialFuture.getNow(absentValue));
+	}
+
+	@Test
+	public void testAccept() throws ExecutionException, InterruptedException {
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+		final AtomicInteger atomicInteger = new AtomicInteger(0);
+		int expectedValue = 42;
+
+		Future<Void> result = initialFuture.thenAcceptAsync(new AcceptFunction<Integer>()
{
+			@Override
+			public void accept(Integer value) {
+				atomicInteger.set(value);
+			}
+		}, executor);
+
+		initialFuture.complete(expectedValue);
+
+		result.get();
+
+		assertEquals(expectedValue, atomicInteger.get());
+	}
+
+	@Test
+	public void testHandle() throws ExecutionException, InterruptedException {
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+		int expectedValue = 43;
+
+		Future<String> result = initialFuture.handleAsync(new BiFunction<Integer, Throwable,
String>() {
+			@Override
+			public String apply(Integer integer, Throwable throwable) {
+				if (integer != null) {
+					return String.valueOf(integer);
+				} else {
+					return throwable.getMessage();
+				}
+			}
+		}, executor);
+
+		initialFuture.complete(expectedValue);
+
+		assertEquals(String.valueOf(expectedValue), result.get());
+	}
+
+	@Test
+	public void testHandleException() throws ExecutionException, InterruptedException {
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+		String exceptionMessage = "foobar";
+
+		Future<String> result = initialFuture.handleAsync(new BiFunction<Integer, Throwable,
String>() {
+			@Override
+			public String apply(Integer integer, Throwable throwable) {
+				if (integer != null) {
+					return String.valueOf(integer);
+				} else {
+					return throwable.getMessage();
+				}
+			}
+		}, executor);
+
+		initialFuture.completeExceptionally(new TestException(exceptionMessage));
+
+		assertEquals(exceptionMessage, result.get());
+	}
+
+	@Test
+	public void testMultipleCompleteOperations() throws ExecutionException, InterruptedException
{
+		CompletableFuture<Integer> initialFuture = new FlinkCompletableFuture<>();
+		int expectedValue = 42;
+
+		assertTrue(initialFuture.complete(expectedValue));
+
+		assertFalse(initialFuture.complete(1337));
+
+		assertFalse(initialFuture.completeExceptionally(new TestException("foobar")));
+
+		assertEquals(new Integer(expectedValue), initialFuture.get());
+	}
+
+	private static class TestException extends RuntimeException {
+
+		private static final long serialVersionUID = -1274022962838535130L;
+
+		public TestException(String message) {
+			super(message);
+		}
+	}
+}


Mime
View raw message