You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/11/16 08:21:12 UTC
[flink] branch release-1.7 updated (6c21335 -> f5db114)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 6c21335 [FLINK-10856] Find latest completed checkpoint for resume from externalized checkpoint e2e test
new a2a23b2 [hotfix] Fix checkstyle violation in ExecutionGraph
new f5db114 [FLINK-10883] Failing batch jobs with NoResourceAvailableException when slot request times out
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/runtime/executiongraph/Execution.java | 14 +++++-
.../runtime/executiongraph/ExecutionGraph.java | 2 +-
.../runtime/minicluster/MiniClusterITCase.java | 56 ++++++++++++++--------
3 files changed, 51 insertions(+), 21 deletions(-)
[flink] 01/02: [hotfix] Fix checkstyle violation in ExecutionGraph
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a2a23b233bfcbcf2c6848314453d4744478ed217
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Nov 15 17:27:09 2018 +0100
[hotfix] Fix checkstyle violation in ExecutionGraph
---
.../java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index 3b55e00..56315e0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -911,7 +911,7 @@ public class ExecutionGraph implements AccessExecutionGraph {
final CompletableFuture<Void> schedulingJobVertexFuture = ejv.scheduleAll(
slotProvider,
allowQueuedScheduling,
- LocationPreferenceConstraint.ALL,// since it is an input vertex, the input based location preferences should be empty
+ LocationPreferenceConstraint.ALL, // since it is an input vertex, the input based location preferences should be empty
Collections.emptySet());
schedulingFutures.add(schedulingJobVertexFuture);
[flink] 02/02: [FLINK-10883] Failing batch jobs with
NoResourceAvailableException when slot request times out
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f5db114940fad1b96b549801c80b312a9367a55a
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Nov 15 17:27:22 2018 +0100
[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.
---
.../flink/runtime/executiongraph/Execution.java | 14 +++++-
.../runtime/minicluster/MiniClusterITCase.java | 56 ++++++++++++++--------
2 files changed, 50 insertions(+), 20 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index cbf5103..e3b501e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -41,6 +41,7 @@ import org.apache.flink.runtime.io.network.ConnectionID;
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 class Execution implements AccessExecution, Archiveable<ArchivedExecution
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);
}
});
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
index c108eae..585e1ba 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
@@ -101,7 +101,42 @@ public class MiniClusterITCase extends TestLogger {
}
@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 class MiniClusterITCase extends TestLogger {
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);
}
}