Repository: flink Updated Branches: refs/heads/master 69843fefc -> ca681101f
[hotfix] [dist. coordination] Clean up exception signature of ExecutionGraph Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/231bec8d Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/231bec8d Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/231bec8d Branch: refs/heads/master Commit: 231bec8d4d5b2e79e279ea1acfd9d4f14f892728 Parents: c277ee1 Author: Stephan Ewen <[email protected]> Authored: Wed Mar 29 17:09:48 2017 +0200 Committer: Stephan Ewen <[email protected]> Committed: Wed Mar 29 17:11:49 2017 +0200 ---------------------------------------------------------------------- .../flink/runtime/executiongraph/ExecutionGraph.java | 14 +++++++++++--- .../runtime/executiongraph/ExecutionGraphBuilder.java | 10 ++-------- 2 files changed, 13 insertions(+), 11 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/231bec8d/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java ---------------------------------------------------------------------- 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 06b2f9a..b21b72b 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 @@ -63,6 +63,7 @@ import org.apache.flink.runtime.state.StateBackend; import org.apache.flink.runtime.taskmanager.TaskExecutionState; import org.apache.flink.runtime.util.SerializedThrowable; import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.FlinkRuntimeException; import org.apache.flink.util.Preconditions; import org.apache.flink.util.SerializedValue; @@ -240,7 +241,7 @@ public class ExecutionGraph implements AccessExecutionGraph, Archiveable<Archive SerializedValue<ExecutionConfig> serializedConfig, Time timeout, RestartStrategy restartStrategy, - SlotProvider slotProvider) throws IOException { + SlotProvider slotProvider) { this( futureExecutor, ioExecutor, @@ -268,7 +269,7 @@ public class ExecutionGraph implements AccessExecutionGraph, Archiveable<Archive List<BlobKey> requiredJarFiles, List<URL> requiredClasspaths, SlotProvider slotProvider, - ClassLoader userClassLoader) throws IOException { + ClassLoader userClassLoader) { checkNotNull(futureExecutor); checkNotNull(jobId); @@ -284,7 +285,14 @@ public class ExecutionGraph implements AccessExecutionGraph, Archiveable<Archive requiredClasspaths); // serialize the job information to do the serialisation work only once - this.serializedJobInformation = new SerializedValue<>(jobInformation); + try { + this.serializedJobInformation = new SerializedValue<>(jobInformation); + } + catch (IOException e) { + // this cannot happen because 'JobInformation' is perfectly serializable + // rethrow unchecked, because this indicates a bug, not a recoverable situation + throw new FlinkRuntimeException("Bug: Cannot serialize JobInformation", e); + } this.futureExecutor = Preconditions.checkNotNull(futureExecutor); this.ioExecutor = Preconditions.checkNotNull(ioExecutor); http://git-wip-us.apache.org/repos/asf/flink/blob/231bec8d/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java index 494b7a2..f1da8bd 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java @@ -87,11 +87,8 @@ public class ExecutionGraphBuilder { final JobID jobId = jobGraph.getJobID(); // create a new execution graph, if none exists so far - final ExecutionGraph executionGraph; - - try { - executionGraph = (prior != null) ? prior : - new ExecutionGraph( + final ExecutionGraph executionGraph = (prior != null) ? prior : + new ExecutionGraph( futureExecutor, ioExecutor, jobId, @@ -104,9 +101,6 @@ public class ExecutionGraphBuilder { jobGraph.getClasspaths(), slotProvider, classLoader); - } catch (IOException e) { - throw new JobException("Could not create the execution graph.", e); - } // set the basic properties
