[FLINK-10883] Failing batch jobs with NoResourceAvailableException when slot request times out

Instead of failing the ExecutionGraph with a generic TimeoutException if a slot request times out,
this commit changes the exception to a more meaningful NoResourceAvailableException.

Author: tillrohrmann

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

@@ -41,6 +41,7 @@
 import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
 import org.apache.flink.runtime.jobmanager.scheduler.CoLocationConstraint;
 import org.apache.flink.runtime.jobmanager.scheduler.LocationPreferenceConstraint;
+import org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException;
 import org.apache.flink.runtime.jobmanager.scheduler.ScheduledUnit;
 import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
 import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
@@ -427,7 +428,18 @@ public CompletableFuture<Void> scheduleForExecution(
 			deploymentFuture.whenComplete(
 				(Void ignored, Throwable failure) -> {
 					if (failure != null) {
-						markFailed(ExceptionUtils.stripCompletionException(failure));
+						final Throwable stripCompletionException = ExceptionUtils.stripCompletionException(failure);
+						final Throwable schedulingFailureCause;
+
+						if (stripCompletionException instanceof TimeoutException) {
+							schedulingFailureCause = new NoResourceAvailableException(
+								"Could not allocate enough slots within timeout of " + allocationTimeout + " to run the job. " +
+									"Please make sure that the cluster has enough resources.");
+						} else {
+							schedulingFailureCause = stripCompletionException;
+						}
+
+						markFailed(schedulingFailureCause);
 					}
 				});
 

flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java

@@ -101,7 +101,42 @@ public void runJobWithMultipleRpcServices() throws Exception {
 	}
 
 	@Test
-	public void testHandleJobsWhenNotEnoughSlot() throws Exception {
+	public void testHandleStreamingJobsWhenNotEnoughSlot() throws Exception {
+		try {
+			setupAndRunHandleJobsWhenNotEnoughSlots(ScheduleMode.EAGER);
+			fail("Job should fail.");
+		} catch (JobExecutionException e) {
+			assertTrue(findThrowableWithMessage(e, "Job execution failed.").isPresent());
+			assertTrue(findThrowable(e, NoResourceAvailableException.class).isPresent());
+			assertTrue(findThrowableWithMessage(e, "Slots required: 2, slots allocated: 1").isPresent());
+		}
+	}
+
+	@Test
+	public void testHandleBatchJobsWhenNotEnoughSlot() throws Exception {
+		try {
+			setupAndRunHandleJobsWhenNotEnoughSlots(ScheduleMode.LAZY_FROM_SOURCES);
+			fail("Job should fail.");
+		} catch (JobExecutionException e) {
+			assertTrue(findThrowableWithMessage(e, "Job execution failed.").isPresent());
+			assertTrue(findThrowable(e, NoResourceAvailableException.class).isPresent());
+			assertTrue(findThrowableWithMessage(e, "Could not allocate enough slots").isPresent());
+		}
+	}
+
+	private void setupAndRunHandleJobsWhenNotEnoughSlots(ScheduleMode scheduleMode) throws Exception {
+		final JobVertex vertex = new JobVertex("Test Vertex");
+		vertex.setParallelism(2);
+		vertex.setMaxParallelism(2);
+		vertex.setInvokableClass(BlockingNoOpInvokable.class);
+
+		final JobGraph jobGraph = new JobGraph("Test Job", vertex);
+		jobGraph.setScheduleMode(scheduleMode);
+
+		runHandleJobsWhenNotEnoughSlots(jobGraph);
+	}
+
+	private void runHandleJobsWhenNotEnoughSlots(final JobGraph jobGraph) throws Exception {
 		final Configuration configuration = getDefaultConfiguration();
 		configuration.setLong(JobManagerOptions.SLOT_REQUEST_TIMEOUT, 100L);
 
@@ -114,24 +149,7 @@ public void testHandleJobsWhenNotEnoughSlot() throws Exception {
 		try (final MiniCluster miniCluster = new MiniCluster(cfg)) {
 			miniCluster.start();
 
-			final JobVertex vertex = new JobVertex("Test Vertex");
-			vertex.setParallelism(2);
-			vertex.setMaxParallelism(2);
-			vertex.setInvokableClass(BlockingNoOpInvokable.class);
-
-			final JobGraph jobGraph = new JobGraph("Test Job", vertex);
-			jobGraph.setScheduleMode(ScheduleMode.EAGER);
-
-			try {
-				miniCluster.executeJobBlocking(jobGraph);
-
-				fail("Job should fail.");
-			} catch (JobExecutionException e) {
-				assertTrue(findThrowableWithMessage(e, "Job execution failed.").isPresent());
-
-				assertTrue(findThrowable(e, NoResourceAvailableException.class).isPresent());
-				assertTrue(findThrowableWithMessage(e, "Slots required: 2, slots allocated: 1").isPresent());
-			}
+			miniCluster.executeJobBlocking(jobGraph);
 		}
 	}