[FLINK-3327] ExecutionConfig to JobGraph

This closes apache#1583

flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java

@@ -59,9 +59,6 @@ public class ExecutionConfig implements Serializable {
 
 	private static final long serialVersionUID = 1L;
 
-	// Key for storing it in the Job Configuration
-	public static final String CONFIG_KEY = "runtime.config";
-
 	/**
 	 * The constant to use for the parallelism, if the system should use the number
 	 *  of currently available slots.
@@ -648,7 +645,8 @@ public boolean equals(Object obj) {
 				Objects.equals(executionMode, other.executionMode) &&
 				useClosureCleaner == other.useClosureCleaner &&
 				parallelism == other.parallelism &&
-				restartStrategyConfiguration.equals(other.restartStrategyConfiguration) &&
+				((restartStrategyConfiguration == null && other.restartStrategyConfiguration == null) ||
+				restartStrategyConfiguration.equals(other.restartStrategyConfiguration)) &&
 				forceKryo == other.forceKryo &&
 				objectReuse == other.objectReuse &&
 				autoTypeRegistrationEnabled == other.autoTypeRegistrationEnabled &&

flink-core/src/main/java/org/apache/flink/api/common/Plan.java

@@ -38,7 +38,6 @@
 import org.apache.flink.api.common.cache.DistributedCache.DistributedCacheEntry;
 import org.apache.flink.api.common.operators.GenericDataSinkBase;
 import org.apache.flink.api.common.operators.Operator;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.util.Visitable;
@@ -294,16 +293,6 @@ public void setDefaultParallelism(int defaultParallelism) {
 		this.defaultParallelism = defaultParallelism;
 	}
 
-	/**
-	 * Returns the specified restart strategy configuration. This configuration defines the used
-	 * restart strategy to be used at runtime.
-	 *
-	 * @return The specified restart strategy configuration
-	 */
-	public RestartStrategies.RestartStrategyConfiguration getRestartStrategyConfiguration() {
-		return getExecutionConfig().getRestartStrategy();
-	}
-
 	/**
 	 * Gets the optimizer post-pass class for this job. The post-pass typically creates utility classes
 	 * for data types and is specific to a particular data model (record, tuple, Scala, ...)

flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java

@@ -20,7 +20,6 @@
 
 import com.fasterxml.jackson.core.JsonFactory;
 
-import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.aggregators.AggregatorRegistry;
 import org.apache.flink.api.common.aggregators.AggregatorWithName;
@@ -81,11 +80,9 @@
 import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
 import org.apache.flink.runtime.operators.util.LocalStrategy;
 import org.apache.flink.runtime.operators.util.TaskConfig;
-import org.apache.flink.util.InstantiationUtil;
 import org.apache.flink.util.StringUtils;
 import org.apache.flink.util.Visitor;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -217,9 +214,7 @@ public JobGraph compileJobGraph(OptimizedPlan program, JobID jobId) {
 		// ----------- finalize the job graph -----------
 
 		// create the job graph object
-		JobGraph graph = new JobGraph(jobId, program.getJobName());
-
-		graph.setRestartStrategyConfiguration(program.getOriginalPlan().getRestartStrategyConfiguration());
+		JobGraph graph = new JobGraph(jobId, program.getJobName(), program.getOriginalPlan().getExecutionConfig());
 		graph.setAllowQueuedScheduling(false);
 		graph.setSessionTimeout(program.getOriginalPlan().getSessionTimeout());
 
@@ -238,15 +233,6 @@ public JobGraph compileJobGraph(OptimizedPlan program, JobID jobId) {
 			DistributedCache.writeFileInfoToConfig(e.getKey(), e.getValue(), graph.getJobConfiguration());
 		}
 
-		try {
-			InstantiationUtil.writeObjectToConfig(
-					program.getOriginalPlan().getExecutionConfig(),
-					graph.getJobConfiguration(),
-					ExecutionConfig.CONFIG_KEY);
-		} catch (IOException e) {
-			throw new RuntimeException("Config object could not be written to Job Configuration: " + e);
-		}
-
 		// release all references again
 		this.vertices = null;
 		this.chainedTasks = null;

flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java

@@ -20,6 +20,7 @@
 
 import akka.actor.ActorSystem;
 import akka.testkit.JavaTestKit;
+import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.memory.MemoryType;
@@ -92,7 +93,7 @@ public void testBackPressuredProducer() throws Exception {
 			final FiniteDuration deadline = new FiniteDuration(60, TimeUnit.SECONDS);
 
 			// The JobGraph
-			final JobGraph jobGraph = new JobGraph();
+			final JobGraph jobGraph = new JobGraph(new ExecutionConfig());
 			final int parallelism = 4;
 
 			final JobVertex task = new JobVertex("Task");

flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/StackTraceSampleCoordinatorITCase.java

@@ -20,6 +20,7 @@
 
 import akka.actor.ActorSystem;
 import akka.testkit.JavaTestKit;
+import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.AkkaUtils;
@@ -76,7 +77,7 @@ public void testTaskClearedWhileSampling() throws Exception {
 			final FiniteDuration deadline = new FiniteDuration(60, TimeUnit.SECONDS);
 
 			// The JobGraph
-			final JobGraph jobGraph = new JobGraph();
+			final JobGraph jobGraph = new JobGraph(new ExecutionConfig());
 			final int parallelism = 1;
 
 			final JobVertex task = new JobVertex("Task");

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.deployment;
 
+import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.TaskInfo;
 import org.apache.flink.configuration.Configuration;
@@ -88,16 +89,19 @@ public final class TaskDeploymentDescriptor implements Serializable {
 
 	private final SerializedValue<StateHandle<?>> operatorState;
 
+	/** The execution configuration (see {@link ExecutionConfig}) related to the specific job. */
+	private final ExecutionConfig executionConfig;
+
 	private long recoveryTimestamp;
 
 	/**
 	 * Constructs a task deployment descriptor.
 	 */
 	public TaskDeploymentDescriptor(
 			JobID jobID, JobVertexID vertexID, ExecutionAttemptID executionId,
-			String taskName, int indexInSubtaskGroup, int numberOfSubtasks, int attemptNumber,
-			Configuration jobConfiguration, Configuration taskConfiguration, String invokableClassName,
-			List<ResultPartitionDeploymentDescriptor> producedPartitions,
+			ExecutionConfig executionConfig, String taskName, int indexInSubtaskGroup, int numberOfSubtasks,
+			int attemptNumber, Configuration jobConfiguration, Configuration taskConfiguration,
+			String invokableClassName, List<ResultPartitionDeploymentDescriptor> producedPartitions,
 			List<InputGateDeploymentDescriptor> inputGates,
 			List<BlobKey> requiredJarFiles, List<URL> requiredClasspaths,
 			int targetSlotNumber, SerializedValue<StateHandle<?>> operatorState,
@@ -111,6 +115,7 @@ public TaskDeploymentDescriptor(
 		this.jobID = checkNotNull(jobID);
 		this.vertexID = checkNotNull(vertexID);
 		this.executionId = checkNotNull(executionId);
+		this.executionConfig = checkNotNull(executionConfig);
 		this.taskName = checkNotNull(taskName);
 		this.indexInSubtaskGroup = indexInSubtaskGroup;
 		this.numberOfSubtasks = numberOfSubtasks;
@@ -129,16 +134,23 @@ public TaskDeploymentDescriptor(
 
 	public TaskDeploymentDescriptor(
 			JobID jobID, JobVertexID vertexID, ExecutionAttemptID executionId,
-			String taskName, int indexInSubtaskGroup, int numberOfSubtasks, int attemptNumber,
-			Configuration jobConfiguration, Configuration taskConfiguration, String invokableClassName,
-			List<ResultPartitionDeploymentDescriptor> producedPartitions,
+			ExecutionConfig executionConfig, String taskName, int indexInSubtaskGroup, int numberOfSubtasks,
+			int attemptNumber, Configuration jobConfiguration, Configuration taskConfiguration,
+			String invokableClassName, List<ResultPartitionDeploymentDescriptor> producedPartitions,
 			List<InputGateDeploymentDescriptor> inputGates,
 			List<BlobKey> requiredJarFiles, List<URL> requiredClasspaths,
 			int targetSlotNumber) {
 
-		this(jobID, vertexID, executionId, taskName, indexInSubtaskGroup, numberOfSubtasks, attemptNumber,
-				jobConfiguration, taskConfiguration, invokableClassName, producedPartitions,
-				inputGates, requiredJarFiles, requiredClasspaths, targetSlotNumber, null, -1);
+		this(jobID, vertexID, executionId, executionConfig, taskName, indexInSubtaskGroup,
+				numberOfSubtasks, attemptNumber, jobConfiguration, taskConfiguration, invokableClassName,
+				producedPartitions, inputGates, requiredJarFiles, requiredClasspaths, targetSlotNumber, null, -1);
+	}
+
+	/**
+	 * Returns the execution configuration (see {@link ExecutionConfig}) related to the specific job.
+	 */
+	public ExecutionConfig getExecutionConfig() {
+		return executionConfig;
 	}
 
 	/**

flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.execution;
 
+import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.TaskInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.fs.Path;
@@ -45,6 +46,13 @@
  */
 public interface Environment {
 
+	/**
+	 * Returns the job specific {@link ExecutionConfig}.
+	 *
+	 * @return The execution configuration associated with the current job.
+	 * */
+	ExecutionConfig getExecutionConfig();
+
 	/**
 	 * Returns the ID of the job that the task belongs to.
 	 *

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

@@ -20,6 +20,7 @@
 
 import akka.actor.ActorSystem;
 
+import com.google.common.base.Preconditions;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.accumulators.Accumulator;
 import org.apache.flink.api.common.accumulators.AccumulatorHelper;
@@ -60,7 +61,6 @@
 import org.apache.flink.runtime.util.SerializedThrowable;
 import org.apache.flink.util.SerializedValue;
 import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.InstantiationUtil;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -180,6 +180,9 @@ public class ExecutionGraph implements Serializable {
 
 	// ------ Configuration of the Execution -------
 
+	/** The execution configuration (see {@link ExecutionConfig}) related to this specific job. */
+	private ExecutionConfig executionConfig;
+
 	/** Flag to indicate whether the scheduler may queue tasks for execution, or needs to be able
 	 * to deploy them immediately. */
 	private boolean allowQueuedScheduling = false;
@@ -234,7 +237,6 @@ public class ExecutionGraph implements Serializable {
 	private ExecutionContext executionContext;
 
 	// ------ Fields that are only relevant for archived execution graphs ------------
-	private ExecutionConfig executionConfig;
 
 	private String jsonPlan;
 
@@ -250,13 +252,15 @@ public class ExecutionGraph implements Serializable {
 			JobID jobId,
 			String jobName,
 			Configuration jobConfig,
+			ExecutionConfig config,
 			FiniteDuration timeout,
 			RestartStrategy restartStrategy) {
 		this(
 			executionContext,
 			jobId,
 			jobName,
 			jobConfig,
+			config,
 			timeout,
 			restartStrategy,
 			new ArrayList<BlobKey>(),
@@ -270,6 +274,7 @@ public ExecutionGraph(
 			JobID jobId,
 			String jobName,
 			Configuration jobConfig,
+			ExecutionConfig config,
 			FiniteDuration timeout,
 			RestartStrategy restartStrategy,
 			List<BlobKey> requiredJarFiles,
@@ -302,7 +307,7 @@ public ExecutionGraph(
 
 		this.requiredJarFiles = requiredJarFiles;
 		this.requiredClasspaths = requiredClasspaths;
-
+		this.executionConfig = Preconditions.checkNotNull(config);
 		this.timeout = timeout;
 
 		this.restartStrategy = restartStrategy;
@@ -942,12 +947,7 @@ public void prepareForArchiving() {
 		if (!state.isTerminalState()) {
 			throw new IllegalStateException("Can only archive the job from a terminal state");
 		}
-		// "unpack" execution config before we throw away the usercode classloader.
-		try {
-			executionConfig = (ExecutionConfig) InstantiationUtil.readObjectFromConfig(jobConfiguration, ExecutionConfig.CONFIG_KEY,userClassLoader);
-		} catch (Exception e) {
-			LOG.warn("Error deserializing the execution config while archiving the execution graph", e);
-		}
+
 		// clear the non-serializable fields
 		userClassLoader = null;
 		scheduler = null;

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

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.executiongraph;
 
+import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.runtime.JobException;
 import org.apache.flink.runtime.blob.BlobKey;
 import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
@@ -665,10 +666,11 @@ TaskDeploymentDescriptor createDeploymentDescriptor(
 			consumedPartitions.add(new InputGateDeploymentDescriptor(resultId, queueToRequest, partitions));
 		}
 
+		ExecutionConfig config = getExecutionGraph().getExecutionConfig();
 		List<BlobKey> jarFiles = getExecutionGraph().getRequiredJarFiles();
 		List<URL> classpaths = getExecutionGraph().getRequiredClasspaths();
 
-		return new TaskDeploymentDescriptor(getJobId(), getJobvertexId(), executionId, getTaskName(),
+		return new TaskDeploymentDescriptor(getJobId(), getJobvertexId(), executionId, config, getTaskName(),
 				subTaskIndex, getTotalNumberOfParallelSubtasks(), attemptNumber, getExecutionGraph().getJobConfiguration(),
 				jobVertex.getJobVertex().getConfiguration(), jobVertex.getJobVertex().getInvokableClassName(),
 				producedPartitions, consumedPartitions, jarFiles, classpaths, targetSlot.getRoot().getSlotNumber(),