[ https://issues.apache.org/jira/browse/FLINK-2111?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14717286#comment-14717286 ]
ASF GitHub Bot commented on FLINK-2111: --------------------------------------- Github user mjsax commented on a diff in the pull request: https://github.com/apache/flink/pull/750#discussion_r38132263 --- Diff: flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala --- @@ -411,6 +411,23 @@ class TaskManager( log.debug(s"Cannot find task to fail for execution ${executionID})") } + // stops a task + case StopTask(executionID) => + val task = runningTasks.get(executionID) + if (task != null) { + try { + task.stopExecution() --- End diff -- Well. We know, that stop is only sent to sources, thus, the user function is of type `SourceFunction`. The `cancel()` method has the following JavaDoc: ``` /** * Cancels the source. Most sources will have a while loop inside the * {@link #run} method. You need to ensure that the source will break out of this loop. This * can be achieved by having a volatile field "isRunning" that is checked in the loop and that * is set to false in this method. */ void cancel(); ``` I interpreted this a a non-blocking function. Even if this requirement is not specified explicitly. We could just do this (need to talk to other streaming folk about implication though). Or we add a new method `stop()` to `SourceFunction`. However, this implies to change a lot of code... All classes implementing `SourceFunction` need to get extended (this was another reason why I just used `cancel()`). > Add "stop" signal to cleanly shutdown streaming jobs > ---------------------------------------------------- > > Key: FLINK-2111 > URL: https://issues.apache.org/jira/browse/FLINK-2111 > Project: Flink > Issue Type: Improvement > Components: Distributed Runtime, JobManager, Local Runtime, > Streaming, TaskManager, Webfrontend > Reporter: Matthias J. Sax > Assignee: Matthias J. Sax > Priority: Minor > > Currently, streaming jobs can only be stopped using "cancel" command, what is > a "hard" stop with no clean shutdown. > The new introduced "stop" signal, will only affect streaming source tasks > such that the sources can stop emitting data and shutdown cleanly, resulting > in a clean shutdown of the whole streaming job. > This feature is a pre-requirment for > https://issues.apache.org/jira/browse/FLINK-1929 -- This message was sent by Atlassian JIRA (v6.3.4#6332)