FLINK-2523: Makes the task cancellation interval configurable.

docs/apis/common/index.md

@@ -999,6 +999,8 @@ Note that types registered with `registerKryoType()` are not available to Flink'
 
 - `disableAutoTypeRegistration()` Automatic type registration is enabled by default. The automatic type registration is registering all types (including sub-types) used by usercode with Kryo and the POJO serializer.
 
+- `setTaskCancellationInterval(long interval)` Sets the the interval (in milliseconds) to wait between consecutive attempts to cancel a running task. When a task is canceled, a new thread is created, which calls periodically `interrupt()` on the task thread, if the task thread does not terminate within a certain time. This parameter refers to the time between consecutive calls to `interrupt()` and is set by default to **30000** milliseconds, or **30 seconds**.
+
 The `RuntimeContext` which is accessible in `Rich*` functions through the `getRuntimeContext()` method also allows to access the `ExecutionConfig` in all user defined functions.
 
 {% top %}

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

@@ -22,6 +22,8 @@
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.ConfigConstants;
+
 
 import java.io.Serializable;
 import java.util.LinkedHashMap;
@@ -59,9 +61,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.
@@ -111,7 +110,9 @@ public class ExecutionConfig implements Serializable {
 	private long executionRetryDelay = DEFAULT_RESTART_DELAY;
 
 	private RestartStrategies.RestartStrategyConfiguration restartStrategyConfiguration;
-
+
+	private long taskCancellationIntervalMillis = ConfigConstants.DEFAULT_TASK_CANCELLATION_INTERVAL_MILLIS;
+
 	// Serializers and types registered with Kryo and the PojoSerializer
 	// we store them in linked maps/sets to ensure they are registered in order in all kryo instances.
 
@@ -218,6 +219,28 @@ public ExecutionConfig setParallelism(int parallelism) {
 		return this;
 	}
 
+	/**
+	 * Gets the interval (in milliseconds) between consecutive attempts to cancel a running task.
+	 */
+	public long getTaskCancellationInterval() {
+		return this.taskCancellationIntervalMillis;
+	}
+
+	/**
+	 * Sets the configuration parameter specifying the interval (in milliseconds)
+	 * between consecutive attempts to cancel a running task.
+	 * @param interval the interval (in milliseconds).
+	 */
+	public ExecutionConfig setTaskCancellationInterval(long interval) {
+		if(interval < 0) {
+			throw new IllegalArgumentException(
+				"The task cancellation interval cannot be negative."
+			);
+		}
+		this.taskCancellationIntervalMillis = interval;
+		return this;
+	}
+
 	/**
 	 * Sets the restart strategy to be used for recovery.
 	 *
@@ -650,7 +673,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 &&
@@ -663,7 +687,8 @@ public boolean equals(Object obj) {
 				registeredTypesWithKryoSerializerClasses.equals(other.registeredTypesWithKryoSerializerClasses) &&
 				defaultKryoSerializerClasses.equals(other.defaultKryoSerializerClasses) &&
 				registeredKryoTypes.equals(other.registeredKryoTypes) &&
-				registeredPojoTypes.equals(other.registeredPojoTypes);
+				registeredPojoTypes.equals(other.registeredPojoTypes) &&
+				taskCancellationIntervalMillis == other.taskCancellationIntervalMillis;
 
 		} else {
 			return false;
@@ -689,7 +714,8 @@ public int hashCode() {
 			registeredTypesWithKryoSerializerClasses,
 			defaultKryoSerializerClasses,
 			registeredKryoTypes,
-			registeredPojoTypes);
+			registeredPojoTypes,
+			taskCancellationIntervalMillis);
 	}
 
 	public boolean canEqual(Object obj) {

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-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java

@@ -660,6 +660,11 @@ public final class ConfigConstants {
 	 */
 	public static final boolean DEFAULT_TASK_MANAGER_MEMORY_PRE_ALLOCATE = false;
 
+	/**
+	 * The default interval (in milliseconds) to wait between consecutive task cancellation attempts (= 30000 msec).
+	 * */
+	public static final long DEFAULT_TASK_CANCELLATION_INTERVAL_MILLIS = 30000;
+
 	// ------------------------ Runtime Algorithms ------------------------
 
 	/**

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;