From 842772d5dde87e93a48c5926eb1e587a09c29220 Mon Sep 17 00:00:00 2001 From: Kostas Kloudas Date: Wed, 20 Jan 2016 14:05:58 +0100 Subject: [PATCH] FLINK-2523: making the task cancellation interval configurable. --- docs/apis/batch/index.md | 2 +- .../flink/api/common/ExecutionConfig.java | 31 +++++++++++++++-- .../flink/configuration/ConfigConstants.java | 5 +++ .../deployment/TaskDeploymentDescriptor.java | 33 +++++++++++++------ .../executiongraph/ExecutionGraph.java | 15 ++++++--- .../executiongraph/ExecutionVertex.java | 2 +- .../flink/runtime/taskmanager/Task.java | 32 ++++++++++++------ .../runtime/taskmanager/TaskManager.scala | 1 - .../flink/runtime/taskmanager/TaskTest.java | 3 +- .../runtime/jobmanager/JobManagerITCase.scala | 2 +- .../api/graph/StreamingJobGraphGenerator.java | 4 +-- 11 files changed, 96 insertions(+), 34 deletions(-) diff --git a/docs/apis/batch/index.md b/docs/apis/batch/index.md index 2009340ca38367..e61d1f49253897 100644 --- a/docs/apis/batch/index.md +++ b/docs/apis/batch/index.md @@ -2124,7 +2124,7 @@ 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. By default this is set 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. diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java index 3e2e2fd2f504d2..ceb4f343d81683 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java @@ -21,6 +21,7 @@ import com.esotericsoftware.kryo.Serializer; import org.apache.flink.annotation.Experimental; import org.apache.flink.annotation.Public; +import org.apache.flink.configuration.ConfigConstants; import java.io.Serializable; import java.util.LinkedHashMap; @@ -99,6 +100,8 @@ public class ExecutionConfig implements Serializable { private long executionRetryDelay = -1; + 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. @@ -244,6 +247,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; + } + /** * Gets the number of times the system will try to re-execute failed tasks. A value * of {@code -1} indicates that the system default value (as defined in the configuration) @@ -627,7 +652,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; @@ -653,7 +679,8 @@ public int hashCode() { registeredTypesWithKryoSerializerClasses, defaultKryoSerializerClasses, registeredKryoTypes, - registeredPojoTypes); + registeredPojoTypes, + taskCancellationIntervalMillis); } public boolean canEqual(Object obj) { diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java index aba3540eaee9be..dc17e3337bf374 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java @@ -601,6 +601,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 ------------------------ /** diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java index 912a0cea13ea62..5e2f790eed6635 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java @@ -19,6 +19,7 @@ package org.apache.flink.runtime.deployment; import org.apache.flink.api.common.ApplicationID; +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; @@ -93,24 +94,28 @@ public final class TaskDeploymentDescriptor implements Serializable { private final SerializedValue> operatorState; private long recoveryTimestamp; - + + /** The interval (in millis) between consecutive task cancellation retries. */ + private final ExecutionConfig executionConfig; + /** * Constructs a task deployment descriptor. */ public TaskDeploymentDescriptor( - ApplicationID appId, JobID jobID, JobVertexID vertexID, ExecutionAttemptID executionId, - String taskName, int indexInSubtaskGroup, int numberOfSubtasks, int attemptNumber, - Configuration jobConfiguration, Configuration taskConfiguration, String invokableClassName, - List producedPartitions, - List inputGates, - List requiredJarFiles, List requiredClasspaths, - int targetSlotNumber, SerializedValue> operatorState, - long recoveryTimestamp) { + ApplicationID appId, JobID jobID, JobVertexID vertexID, ExecutionAttemptID executionId, + String taskName, int indexInSubtaskGroup, int numberOfSubtasks, int attemptNumber, + Configuration jobConfiguration, Configuration taskConfiguration, String invokableClassName, + List producedPartitions, + List inputGates, + List requiredJarFiles, List requiredClasspaths, + int targetSlotNumber, SerializedValue> operatorState, + long recoveryTimestamp, ExecutionConfig executionConfig) { checkArgument(indexInSubtaskGroup >= 0); checkArgument(numberOfSubtasks > indexInSubtaskGroup); checkArgument(targetSlotNumber >= 0); checkArgument(attemptNumber >= 0); + checkArgument(executionConfig != null); this.appId = checkNotNull(appId); this.jobID = checkNotNull(jobID); @@ -130,6 +135,7 @@ public TaskDeploymentDescriptor( this.targetSlotNumber = targetSlotNumber; this.operatorState = operatorState; this.recoveryTimestamp = recoveryTimestamp; + this.executionConfig = executionConfig; } public TaskDeploymentDescriptor( @@ -143,7 +149,7 @@ public TaskDeploymentDescriptor( this(appId, jobID, vertexID, executionId, taskName, indexInSubtaskGroup, numberOfSubtasks, attemptNumber, jobConfiguration, taskConfiguration, invokableClassName, producedPartitions, - inputGates, requiredJarFiles, requiredClasspaths, targetSlotNumber, null, -1); + inputGates, requiredJarFiles, requiredClasspaths, targetSlotNumber, null, -1, new ExecutionConfig()); } /** @@ -208,6 +214,13 @@ public TaskInfo getTaskInfo() { return new TaskInfo(taskName, indexInSubtaskGroup, numberOfSubtasks, attemptNumber); } + /** + * Returns the execution configuration of the job at hand. + */ + public ExecutionConfig getExecutionConfig() { + return executionConfig; + } + /** * Gets the number of the slot into which the task is to be deployed. * 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 90854836b9378d..6c0a5bb4cb7421 100755 --- 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 @@ -928,11 +928,8 @@ public void prepareForArchiving() { 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); - } + executionConfig = this.getExecutionConfig(); + // clear the non-serializable fields userClassLoader = null; scheduler = null; @@ -953,6 +950,14 @@ public void prepareForArchiving() { } public ExecutionConfig getExecutionConfig() { + if (this.executionConfig == null) { + try { + this.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); + } + } return this.executionConfig; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java index c89a01e623edb1..5a7a04140910f3 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java @@ -654,7 +654,7 @@ TaskDeploymentDescriptor createDeploymentDescriptor( subTaskIndex, getTotalNumberOfParallelSubtasks(), attemptNumber, getExecutionGraph().getJobConfiguration(), jobVertex.getJobVertex().getConfiguration(), jobVertex.getJobVertex().getInvokableClassName(), producedPartitions, consumedPartitions, jarFiles, classpaths, targetSlot.getRoot().getSlotNumber(), - operatorState, recoveryTimestamp); + operatorState, recoveryTimestamp, this.getExecutionGraph().getExecutionConfig()); } // -------------------------------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java index 9cc1be4f03eba4..9e3629b96fec8c 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java @@ -219,6 +219,12 @@ public class Task implements Runnable { private volatile long recoveryTs; + /** + * The interval (in milliseconds) between consecutive attempts to + * cancel a running task (see {@link TaskCanceler}). + * */ + private volatile long cancellationInterval; + /** *

IMPORTANT: This constructor may not start any work that would need to * be undone in the case of a failing task deployment.

@@ -248,6 +254,7 @@ public Task(TaskDeploymentDescriptor tdd, this.nameOfInvokableClass = checkNotNull(tdd.getInvokableClassName()); this.operatorState = tdd.getOperatorState(); this.recoveryTs = tdd.getRecoveryTimestamp(); + this.cancellationInterval = tdd.getExecutionConfig().getTaskCancellationInterval(); this.memoryManager = checkNotNull(memManager); this.ioManager = checkNotNull(ioManager); @@ -811,7 +818,9 @@ else if (current == ExecutionState.RUNNING) { // because the canceling may block on user code, we cancel from a separate thread // we do not reuse the async call handler, because that one may be blocked, in which // case the canceling could not continue - Runnable canceler = new TaskCanceler(LOG, invokable, executingThread, taskNameWithSubtask); + Runnable canceler = new TaskCanceler(LOG, invokable, executingThread, + taskNameWithSubtask, cancellationInterval); + Thread cancelThread = new Thread(executingThread.getThreadGroup(), canceler, "Canceler for " + taskNameWithSubtask); cancelThread.setDaemon(true); @@ -1045,14 +1054,17 @@ private static class TaskCanceler implements Runnable { private final Logger logger; private final AbstractInvokable invokable; - private final Thread executer; + private final Thread executor; private final String taskName; + private final long cancellationIntervalMillis; - public TaskCanceler(Logger logger, AbstractInvokable invokable, Thread executer, String taskName) { + public TaskCanceler(Logger logger, AbstractInvokable invokable, + Thread executor, String taskName, long cancellationInterval) { this.logger = logger; this.invokable = invokable; - this.executer = executer; + this.executor = executor; this.taskName = taskName; + this.cancellationIntervalMillis = cancellationInterval; } @Override @@ -1068,9 +1080,9 @@ public void run() { } // interrupt the running thread initially - executer.interrupt(); + executor.interrupt(); try { - executer.join(30000); + executor.join(this.cancellationIntervalMillis); } catch (InterruptedException e) { // we can ignore this @@ -1079,11 +1091,11 @@ public void run() { // it is possible that the user code does not react immediately. for that // reason, we spawn a separate thread that repeatedly interrupts the user code until // it exits - while (executer.isAlive()) { + while (executor.isAlive()) { // build the stack trace of where the thread is stuck, for the log StringBuilder bld = new StringBuilder(); - StackTraceElement[] stack = executer.getStackTrace(); + StackTraceElement[] stack = executor.getStackTrace(); for (StackTraceElement e : stack) { bld.append(e).append('\n'); } @@ -1091,9 +1103,9 @@ public void run() { logger.warn("Task '{}' did not react to cancelling signal, but is stuck in method:\n {}", taskName, bld.toString()); - executer.interrupt(); + executor.interrupt(); try { - executer.join(30000); + executor.join(this.cancellationIntervalMillis); } catch (InterruptedException e) { // we can ignore this diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala index afceaf986d8478..e5378a4629df27 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala @@ -173,7 +173,6 @@ class TaskManager( var leaderSessionID: Option[UUID] = None - private val runtimeInfo = new TaskManagerRuntimeInfo( connectionInfo.getHostname(), new UnmodifiableConfiguration(config.configuration)) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java index 124fe4ead1517e..69f87866fbb738 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java @@ -21,6 +21,7 @@ import com.google.common.collect.Maps; import org.apache.flink.api.common.ApplicationID; +import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.blob.BlobKey; import org.apache.flink.runtime.broadcast.BroadcastVariableManager; @@ -742,7 +743,7 @@ private TaskDeploymentDescriptor createTaskDeploymentDescriptor(ClassemptyList(), Collections.emptyList(), Collections.emptyList(), - 0); + 0, null, -1, new ExecutionConfig()); } // ------------------------------------------------------------------------ diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala index ec54b7e3f4f0c6..b2bbe00e3db367 100644 --- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala +++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala @@ -126,7 +126,7 @@ class JobManagerITCase(_system: ActorSystem) jmGateway.tell(SubmitJob(jobGraph, ListeningBehaviour.EXECUTION_RESULT), self) expectMsg(JobSubmitSuccess(jobGraph.getJobID)) - + val result = expectMsgType[JobResultSuccess] result.result.getJobId() should equal(jobGraph.getJobID) } diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java index ad96cbf1d84408..1a5ab1c151d148 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java @@ -229,14 +229,14 @@ private List createChain( return transitiveOutEdges; } else { - return new ArrayList(); + return new ArrayList<>(); } } private String createChainedName(Integer vertexID, List chainedOutputs) { String operatorName = streamGraph.getStreamNode(vertexID).getOperatorName(); if (chainedOutputs.size() > 1) { - List outputChainedNames = new ArrayList(); + List outputChainedNames = new ArrayList<>(); for (StreamEdge chainable : chainedOutputs) { outputChainedNames.add(chainedNames.get(chainable.getTargetId())); }