[FLINK-5747] [distributed coordination] Eager scheduling allocates sl…

…ots and deploys tasks in bulk

That way, strictly topological deployment can be guaranteed.

Also, many quick deploy/not-enough-resources/fail/recover cycles can be
avoided in the cases where resources need some time to appear.

This closes apache#3295

flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java

@@ -100,6 +100,18 @@ public static boolean isJvmFatalOrOutOfMemoryError(Throwable t) {
 		return isJvmFatalError(t) || t instanceof OutOfMemoryError;
 	}
 
+	/**
+	 * Rethrows the given {@code Throwable}, if it represents an error that is fatal to the JVM.
+	 * See {@link ExceptionUtils#isJvmFatalError(Throwable)} for a definition of fatal errors.
+	 * 
+	 * @param t The Throwable to check and rethrow.
+	 */
+	public static void rethrowIfFatalError(Throwable t) {
+		if (isJvmFatalError(t)) {
+			throw (Error) t;
+		}
+	}
+
 	/**
 	 * Adds a new exception as a {@link Throwable#addSuppressed(Throwable) suppressed exception}
 	 * to a prior exception, or returns the new exception, if no prior exception exists.

flink-core/src/test/java/org/apache/flink/util/ExceptionUtilsTest.java

@@ -0,0 +1,60 @@
+/*
+ * 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.util;
+
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+/**
+ * Tests for the utility methods in {@link ExceptionUtils}.
+ */
+public class ExceptionUtilsTest {
+
+	@Test
+	public void testStringifyNullException() {
+		assertNotNull(ExceptionUtils.STRINGIFIED_NULL_EXCEPTION);
+		assertEquals(ExceptionUtils.STRINGIFIED_NULL_EXCEPTION, ExceptionUtils.stringifyException(null));
+	}
+
+	@Test
+	public void testJvmFatalError() {
+		// not all errors are fatal
+		assertFalse(ExceptionUtils.isJvmFatalError(new Error()));
+
+		// linkage errors are not fatal
+		assertFalse(ExceptionUtils.isJvmFatalError(new LinkageError()));
+
+		// some errors are fatal
+		assertTrue(ExceptionUtils.isJvmFatalError(new InternalError()));
+		assertTrue(ExceptionUtils.isJvmFatalError(new UnknownError()));
+	}
+
+	@Test
+	public void testRethrowFatalError() {
+		// fatal error is rethrown
+		try {
+			ExceptionUtils.rethrowIfFatalError(new InternalError());
+			fail();
+		} catch (InternalError ignored) {}
+
+		// non-fatal error is not rethrown
+		ExceptionUtils.rethrowIfFatalError(new NoClassDefFoundError());
+	}
+}

flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/Executors.java

@@ -21,6 +21,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nonnull;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -52,7 +53,7 @@ private static class DirectExecutor implements Executor {
 		private DirectExecutor() {}
 
 		@Override
-		public void execute(Runnable command) {
+		public void execute(@Nonnull Runnable command) {
 			command.run();
 		}
 	}

flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/FutureUtils.java

@@ -20,11 +20,22 @@
 
 import org.apache.flink.runtime.concurrent.impl.FlinkCompletableFuture;
 
+import java.util.Collection;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A collection of utilities that expand the usage of {@link Future} and {@link CompletableFuture}.
+ */
 public class FutureUtils {
 
+	// ------------------------------------------------------------------------
+	//  retrying operations
+	// ------------------------------------------------------------------------
+
 	/**
 	 * Retry the given operation the given number of times in case of a failure.
 	 *
@@ -88,4 +99,108 @@ public RetryException(Throwable cause) {
 			super(cause);
 		}
 	}
+
+	// ------------------------------------------------------------------------
+	//  composing futures
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Creates a future that is complete once multiple other futures completed. 
+	 * The ConjunctFuture fails (completes exceptionally) once one of the Futures in the
+	 * conjunction fails.
+	 *
+	 * <p>The ConjunctFuture gives access to how many Futures in the conjunction have already
+	 * completed successfully, via {@link ConjunctFuture#getNumFuturesCompleted()}. 
+	 * 
+	 * @param futures The futures that make up the conjunction. No null entries are allowed.
+	 * @return The ConjunctFuture that completes once all given futures are complete (or one fails).
+	 */
+	public static ConjunctFuture combineAll(Collection<? extends Future<?>> futures) {
+		checkNotNull(futures, "futures");
+
+		final ConjunctFutureImpl conjunct = new ConjunctFutureImpl(futures.size());
+
+		if (futures.isEmpty()) {
+			conjunct.complete(null);
+		}
+		else {
+			for (Future<?> future : futures) {
+				future.handle(conjunct.completionHandler);
+			}
+		}
+
+		return conjunct;
+	}
+
+	/**
+	 * A future that is complete once multiple other futures completed. The futures are not
+	 * necessarily of the same type, which is why the type of this Future is {@code Void}.
+	 * The ConjunctFuture fails (completes exceptionally) once one of the Futures in the
+	 * conjunction fails.
+	 * 
+	 * <p>The advantage of using the ConjunctFuture over chaining all the futures (such as via
+	 * {@link Future#thenCombine(Future, BiFunction)}) is that ConjunctFuture also tracks how
+	 * many of the Futures are already complete.
+	 */
+	public interface ConjunctFuture extends CompletableFuture<Void> {
+
+		/**
+		 * Gets the total number of Futures in the conjunction.
+		 * @return The total number of Futures in the conjunction.
+		 */
+		int getNumFuturesTotal();
+
+		/**
+		 * Gets the number of Futures in the conjunction that are already complete.
+		 * @return The number of Futures in the conjunction that are already complete
+		 */
+		int getNumFuturesCompleted();
+	}
+
+	/**
+	 * The implementation of the {@link ConjunctFuture}.
+	 * 
+	 * <p>Implementation notice: The member fields all have package-private access, because they are
+	 * either accessed by an inner subclass or by the enclosing class.
+	 */
+	private static class ConjunctFutureImpl extends FlinkCompletableFuture<Void> implements ConjunctFuture {
+
+		/** The total number of futures in the conjunction */
+		final int numTotal;
+
+		/** The number of futures in the conjunction that are already complete */
+		final AtomicInteger numCompleted = new AtomicInteger();
+
+		/** The function that is attached to all futures in the conjunction. Once a future
+		 * is complete, this function tracks the completion or fails the conjunct.  
+		 */
+		final BiFunction<Object, Throwable, Void> completionHandler = new BiFunction<Object, Throwable, Void>() {
+
+			@Override
+			public Void apply(Object o, Throwable throwable) {
+				if (throwable != null) {
+					completeExceptionally(throwable);
+				}
+				else if (numTotal == numCompleted.incrementAndGet()) {
+					complete(null);
+				}
+
+				return null;
+			}
+		};
+
+		ConjunctFutureImpl(int numTotal) {
+			this.numTotal = numTotal;
+		}
+
+		@Override
+		public int getNumFuturesTotal() {
+			return numTotal;
+		}
+
+		@Override
+		public int getNumFuturesCompleted() {
+			return numCompleted.get();
+		}
+	}
 }

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java

@@ -249,27 +249,8 @@ public void setInitialState(TaskStateHandles checkpointStateHandles) {
 	 * @throws IllegalStateException Thrown, if the vertex is not in CREATED state, which is the only state that permits scheduling.
 	 */
 	public boolean scheduleForExecution(SlotProvider slotProvider, boolean queued) {
-		if (slotProvider == null) {
-			throw new IllegalArgumentException("Cannot send null Scheduler when scheduling execution.");
-		}
-
-		final SlotSharingGroup sharingGroup = vertex.getJobVertex().getSlotSharingGroup();
-		final CoLocationConstraint locationConstraint = vertex.getLocationConstraint();
-
-		// sanity check
-		if (locationConstraint != null && sharingGroup == null) {
-			throw new RuntimeException("Trying to schedule with co-location constraint but without slot sharing allowed.");
-		}
-
-		if (transitionState(CREATED, SCHEDULED)) {
-
-			ScheduledUnit toSchedule = locationConstraint == null ?
-				new ScheduledUnit(this, sharingGroup) :
-				new ScheduledUnit(this, sharingGroup, locationConstraint);
-
-			// IMPORTANT: To prevent leaks of cluster resources, we need to make sure that slots are returned
-			//     in all cases where the deployment failed. we use many try {} finally {} clauses to assure that
-			final Future<SimpleSlot> slotAllocationFuture = slotProvider.allocateSlot(toSchedule, queued);
+		try {
+			final Future<SimpleSlot> slotAllocationFuture = allocateSlotForExecution(slotProvider, queued);
 
 			// IMPORTANT: We have to use the synchronous handle operation (direct executor) here so
 			// that we directly deploy the tasks if the slot allocation future is completed. This is
@@ -296,28 +277,54 @@ public Void apply(SimpleSlot simpleSlot, Throwable throwable) {
 			});
 
 			// if tasks have to scheduled immediately check that the task has been deployed
-			// TODO: This might be problematic if the future is not completed right away
-			if (!queued) {
-				if (!deploymentFuture.isDone()) {
-					markFailed(new IllegalArgumentException("The slot allocation future has not been completed yet."));
-				}
+			if (!queued && !deploymentFuture.isDone()) {
+				markFailed(new IllegalArgumentException("The slot allocation future has not been completed yet."));
 			}
-			
+
 			return true;
 		}
+		catch (IllegalExecutionStateException e) {
+			return false;
+		}
+	}
+
+	public Future<SimpleSlot> allocateSlotForExecution(SlotProvider slotProvider, boolean queued) 
+			throws IllegalExecutionStateException {
+
+		checkNotNull(slotProvider);
+
+		final SlotSharingGroup sharingGroup = vertex.getJobVertex().getSlotSharingGroup();
+		final CoLocationConstraint locationConstraint = vertex.getLocationConstraint();
+
+		// sanity check
+		if (locationConstraint != null && sharingGroup == null) {
+			throw new IllegalStateException(
+					"Trying to schedule with co-location constraint but without slot sharing allowed.");
+		}
+
+		// this method only works if the execution is in the state 'CREATED'
+		if (transitionState(CREATED, SCHEDULED)) {
+
+			ScheduledUnit toSchedule = locationConstraint == null ?
+					new ScheduledUnit(this, sharingGroup) :
+					new ScheduledUnit(this, sharingGroup, locationConstraint);
+
+			return slotProvider.allocateSlot(toSchedule, queued);
+		}
 		else {
 			// call race, already deployed, or already done
-			return false;
+			throw new IllegalExecutionStateException(this, CREATED, state);
 		}
 	}
 
 	public void deployToSlot(final SimpleSlot slot) throws JobException {
-		// sanity checks
-		if (slot == null) {
-			throw new NullPointerException();
-		}
+		checkNotNull(slot);
+
+		// Check if the TaskManager died in the meantime
+		// This only speeds up the response to TaskManagers failing concurrently to deployments.
+		// The more general check is the timeout of the deployment call
 		if (!slot.isAlive()) {
-			throw new JobException("Target slot for deployment is not alive.");
+			throw new JobException("Target slot (TaskManager) for deployment is no longer alive.");
 		}
 
 		// make sure exactly one deployment call happens from the correct state

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionAndSlot.java

@@ -0,0 +1,46 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.runtime.concurrent.Future;
+import org.apache.flink.runtime.instance.SimpleSlot;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A pair of an {@link Execution} together with a slot future.
+ */
+public class ExecutionAndSlot {
+
+	public final Execution executionAttempt;
+
+	public final Future<SimpleSlot> slotFuture;
+
+	public ExecutionAndSlot(Execution executionAttempt, Future<SimpleSlot> slotFuture) {
+		this.executionAttempt = checkNotNull(executionAttempt);
+		this.slotFuture = checkNotNull(slotFuture);
+	}
+
+	// -----------------------------------------------------------------------
+
+	@Override
+	public String toString() {
+		return super.toString();
+	}
+}