You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/06/19 07:33:17 UTC
[GitHub] [flink] rkhachatryan opened a new pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
rkhachatryan opened a new pull request #12714:
URL: https://github.com/apache/flink/pull/12714
## What is the purpose of the change
Wait for the `LegacySourceFunctionThread` to finish when cancelling the task.
This prevents premature freeing of resources, particularly LibraryCache (see FLINK-15467 discussion).
Implemented by adding `Task. invokableCancelFuture` instead of just joining the source thread in `cancelTask`. This is because `cancelTask` can be called from TaskCanceller, while resources are freed by `Task.executingThread` or upon its death.
## Brief change log
There are three commits that should be squashed after reviewing:
- Return future from `StreamTask.cancelTask` and finishTask - only change of signatures except for `SourceStreamTask`
- Return completion future from `AbstractInvokable` cancel - only change of signatures except for `StreamTask`
- Wait in `Task.cancelInvokable` for completion
## Verifying this change
- Added unit test: `TaskTest.testTaskTerminatesOnlyAfterInvokable`
- Manually tested on cluster
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5d78f3b6765a029761a3d7abcea9ba71ccd73e6e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806)
* 35c875a260b4dca9cd86a67356fa9488f2ea5aae Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825)
* 2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r443460354
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -871,6 +901,9 @@ public void testTerminationFutureCompletesOnNormalExecution() throws Exception {
assertFalse(task.getTerminationFuture().isDone());
+ task.getExecutingThread().join(500); // allow some arbitrary delay for potential termination and re-check
+ assertFalse(task.getTerminationFuture().isDone());
Review comment:
Removed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r442803046
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -819,6 +823,8 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
}
finally {
try {
+ waitInvokableCancelCompletion();
Review comment:
Could you add a test for clean shutdown case as well? (to cover for the bug in the previous version that I commented in the PR)
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -657,20 +657,25 @@ protected void cleanUpInvoke() throws Exception {
}
@Override
- public final void cancel() throws Exception {
+ public final CompletableFuture<Void> cancel() {
isRunning = false;
canceled = true;
// the "cancel task" call must come first, but the cancelables must be
// closed no matter what
- try {
- cancelTask();
- }
- finally {
- mailboxProcessor.allActionsCompleted();
- cancelables.close();
+ return FutureUtils
+ .completedVoidFuture()
+ .thenRun(this::cancelTask)
Review comment:
> Leaving client code to deal with both method throwing an exception and Future completed exceptionally isn't very nice IMO.
That would be an argument if not for the fact that exceptions are (and still should be probably) handled via the caller anyway.
But I guess you are right. To avoid code duplication with try/catch block, you would have to extract the `whenComplete(...)` content to a separate method. This could be argued which is a better/easier to understand way, so if you want let's leave it as it is.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r442789830
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -810,6 +817,7 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
}
// else fall through the loop and
}
+ waitInvokableCancelCompletion();
Review comment:
Yes, we do! I'll fix it, thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r444361745
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
> What's the problem with waiting on the latch as I proposed?
For how long should the test thread wait on the latch (assuming `SourceStreamTask.sourceThread` never exits)?
Or did I misunderstood your proposal?
> It should provide automated test coverage for that bug.
It provides only partial coverage because it doesn't cover integration with `Task`, `StreamTask` and the code that actually unloads the classes. So in terms of coverage, it's equivalent to a set of unit tests.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r442789386
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -861,6 +869,43 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
}
}
+ private void waitInvokableCancelCompletion() throws java.util.concurrent.ExecutionException, java.util.concurrent.TimeoutException {
+ if (!invokableHasBeenCanceled.get()) {
+ return;
+ }
+
+ // set deadline if taskCancellationTimeout is set to have time for cleanup (before being killing by TaskCancelerWatchDog) (best effort)
+ Optional<Deadline> deadline = Optional
+ .of((long) (taskCancellationTimeout * .9))
+ .filter(t -> t > 0)
+ .map(t -> Deadline.fromNow(Duration.ofMillis((t))));
Review comment:
The motivation was to preserve the *current* behaviour and have time for cleanup which includes:
```
releaseResources();
memoryManager.releaseAll(invokable);
fileCache.releaseJob(jobId, executionId);
FileSystemSafetyNet.closeSafetyNetAndGuardedResourcesForThread();
notifyFinalState();
```
However, I see your point (especially given that old behavior didn't have a thread leak issue) so I'll remove the deadline.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 35c875a260b4dca9cd86a67356fa9488f2ea5aae Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825)
* 2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646487825
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 5d78f3b6765a029761a3d7abcea9ba71ccd73e6e (Fri Jun 19 07:35:35 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
}, {
"hash" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906",
"triggerID" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"triggerType" : "PUSH"
}, {
"hash" : "0d85f9ed9df65e13533442c95cc904ee1093aa16",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3965",
"triggerID" : "0d85f9ed9df65e13533442c95cc904ee1093aa16",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0d85f9ed9df65e13533442c95cc904ee1093aa16 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3965)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
}, {
"hash" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833)
* eaf54b89a544a57afc928a689c14e40903c1f73d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5d78f3b6765a029761a3d7abcea9ba71ccd73e6e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806)
* 35c875a260b4dca9cd86a67356fa9488f2ea5aae Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5d78f3b6765a029761a3d7abcea9ba71ccd73e6e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r442789595
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -861,6 +869,43 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
}
}
+ private void waitInvokableCancelCompletion() throws java.util.concurrent.ExecutionException, java.util.concurrent.TimeoutException {
+ if (!invokableHasBeenCanceled.get()) {
+ return;
+ }
+
+ // set deadline if taskCancellationTimeout is set to have time for cleanup (before being killing by TaskCancelerWatchDog) (best effort)
+ Optional<Deadline> deadline = Optional
+ .of((long) (taskCancellationTimeout * .9))
+ .filter(t -> t > 0)
+ .map(t -> Deadline.fromNow(Duration.ofMillis((t))));
+
+ if (taskCancellationTimeout > 0 && !deadline.isPresent()) {
+ LOG.debug("Skip waiting for invokable cancellation to have time for cleanup (taskCancellationTimeout is too small)");
+ return;
+ }
+ LOG.debug("Waiting for invokable cancellation (deadline: {})", deadline);
+
+ // prevent interrupts by TaskInterrupter and other canceling threads (best effort)
+ invokable.setShouldInterruptOnCancel(false);
Review comment:
Right. Ok, I'll remove it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r442952328
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -871,6 +901,9 @@ public void testTerminationFutureCompletesOnNormalExecution() throws Exception {
assertFalse(task.getTerminationFuture().isDone());
+ task.getExecutingThread().join(500); // allow some arbitrary delay for potential termination and re-check
+ assertFalse(task.getTerminationFuture().isDone());
Review comment:
I'm not sure if I understand this part. Was this 500ms delay inducing a failure in this test in the previous version if this PR? If not, I don't think adding such sleep would be justified.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
Is it covering for the user reported bug? It's not using `SourceStreamTask` anywhere, so it looks like it's not? If that's the case, couldn't we add a test in `SourceStreamTaskTest`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5d78f3b6765a029761a3d7abcea9ba71ccd73e6e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
}, {
"hash" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906",
"triggerID" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833)
* eaf54b89a544a57afc928a689c14e40903c1f73d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r458176824
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
I meant that source thread, running mock source, would be waiting on the latch (with some timeout), not the test thread. Test thread would:
1. trigger cancellation
2. maybe wait 10ms
3. assert source task's thread hasn't completed
4. release the latch
5. assert that source task has completed (with a timeout)
> It provides only partial coverage because it doesn't cover integration with Task, StreamTask and the code that actually unloads the classes. So in terms of coverage, it's equivalent to a set of unit tests.
It would provide a coverage for task thread shutting down prematurely, before source thread completes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5d78f3b6765a029761a3d7abcea9ba71ccd73e6e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r443499682
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
Hmm.
> extracted SourceStreamTask.LegacySourceFunctionRunner interface in a preceding hotfix commit.
this commit is not super pretty, maybe we could make it a bit better, but I'm not sure if that's the right direction.
> SourceStreamTaskTest.testWatForCancelCompletion
this doesn't seem to be testing much, it's almost a no-op test for passing the returned value of one getter. It's definitely not testing for the reported bug.
What about creating a mock source that waits on a latch in a loop ignoring interrupts and the test thread fires cancellation & ensures task is not completed? Setup should be quite easy like in `SourceStreamTaskTest#finishingIgnoresExceptions`? What do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r458176824
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
I meant that source thread, running mock source, would be waiting on the latch (with some timeout), not the test thread. Test thread would:
1. trigger cancellation
2. maybe wait 10ms
3. assert source task's thread hasn't completed
4 release the latch
5. assert that source task has completed (with a timeout)
> It provides only partial coverage because it doesn't cover integration with Task, StreamTask and the code that actually unloads the classes. So in terms of coverage, it's equivalent to a set of unit tests.
It would provide a coverage for task thread shutting down prematurely, before source thread completes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r443461641
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
Added `SourceStreamTaskTest.testWatForCancelCompletion`.
For that, I extracted `SourceStreamTask.LegacySourceFunctionRunner` interface in a preceding hotfix commit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
}, {
"hash" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906",
"triggerID" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eaf54b89a544a57afc928a689c14e40903c1f73d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r444319566
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
Yes, e2e test would be an overkill, but semi integration test in `SourceStreamTask` seems to be quite easy.
> it's not possible to reliably test that a program does not perform some action if there is no time boundary (in this case test that task never completes if source thread never completes).
What's the problem with waiting on the latch as I proposed? It should provide automated test coverage for that bug.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r442749136
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -810,6 +817,7 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
}
// else fall through the loop and
}
+ waitInvokableCancelCompletion();
Review comment:
Why is it placed only in the `catch(...)` block? Don't we want to wait in case of clean shutdown or clean completion?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -861,6 +869,43 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
}
}
+ private void waitInvokableCancelCompletion() throws java.util.concurrent.ExecutionException, java.util.concurrent.TimeoutException {
+ if (!invokableHasBeenCanceled.get()) {
+ return;
+ }
+
+ // set deadline if taskCancellationTimeout is set to have time for cleanup (before being killing by TaskCancelerWatchDog) (best effort)
+ Optional<Deadline> deadline = Optional
+ .of((long) (taskCancellationTimeout * .9))
+ .filter(t -> t > 0)
+ .map(t -> Deadline.fromNow(Duration.ofMillis((t))));
+
+ if (taskCancellationTimeout > 0 && !deadline.isPresent()) {
+ LOG.debug("Skip waiting for invokable cancellation to have time for cleanup (taskCancellationTimeout is too small)");
+ return;
+ }
+ LOG.debug("Waiting for invokable cancellation (deadline: {})", deadline);
+
+ // prevent interrupts by TaskInterrupter and other canceling threads (best effort)
+ invokable.setShouldInterruptOnCancel(false);
Review comment:
This change would be perfectly fine without this line right? In that case I would prefer to remove it, I think it would make the code cleaner/easier to understand.
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -430,8 +430,8 @@ protected void processInput(MailboxDefaultAction.Controller controller) throws E
protected void cleanup() {}
@Override
- protected void cancelTask() throws Exception {
- throw new Exception("test exception");
+ protected CompletableFuture<Void> cancelTask() {
+ throw new RuntimeException("test exception");
Review comment:
nit: `ExpectedTestException`?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -657,20 +657,25 @@ protected void cleanUpInvoke() throws Exception {
}
@Override
- public final void cancel() throws Exception {
+ public final CompletableFuture<Void> cancel() {
isRunning = false;
canceled = true;
// the "cancel task" call must come first, but the cancelables must be
// closed no matter what
- try {
- cancelTask();
- }
- finally {
- mailboxProcessor.allActionsCompleted();
- cancelables.close();
+ return FutureUtils
+ .completedVoidFuture()
+ .thenRun(this::cancelTask)
Review comment:
Why are you using this construct? With `FutureUtils.completedVoidFuture()`? Why not
```
this.cancelTask().whenComplete(...);
```
+ eventually some exception handling if `cancelTask()` throws.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -861,6 +869,43 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
}
}
+ private void waitInvokableCancelCompletion() throws java.util.concurrent.ExecutionException, java.util.concurrent.TimeoutException {
+ if (!invokableHasBeenCanceled.get()) {
+ return;
+ }
+
+ // set deadline if taskCancellationTimeout is set to have time for cleanup (before being killing by TaskCancelerWatchDog) (best effort)
+ Optional<Deadline> deadline = Optional
+ .of((long) (taskCancellationTimeout * .9))
+ .filter(t -> t > 0)
+ .map(t -> Deadline.fromNow(Duration.ofMillis((t))));
Review comment:
What's the motivation behind this deadline? I think it should be perfectly fine/safe to simplify this and just wait on the future indefinitely, as this is the behaviour for network tasks and for source tasks before we introduced the mailbox (in the past, `SourceFunction#run()` loop was called from the task thread).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r443686425
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
> It's definitely not testing for the reported bug.
A proper test for the reported bug would be an end-to-end test involving running a remote cluster, syncing source thread with class unloading and then checking the logs.
I think we agreed offline that such automated test would be an overkill and could rather be done manually. And the rest should be covered with unit tests.
Now, this unit test checks that `SourceStreamTask` returns completion future of its thread.
This is what changed - and this is what I think **should** be tested.
If we also want to test `LegacySourceFunctionThread` future completion, it can be done independently now after extraction (however, I think this is out of the scope of this PR).
Furthermore, it's not possible to reliably test that a program does not perform some action if there is no time boundary (in this case test that task never completes if source thread never completes).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r444319566
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -139,6 +142,33 @@ public void testRegularExecution() throws Exception {
taskManagerActions.validateListenerMessage(ExecutionState.FINISHED, task, null);
}
+ @Test
+ public void testTaskTerminatesOnlyAfterInvokable() throws Exception {
Review comment:
Yes, e2e test would be an overkill, but semi integration test in `SourceStreamTask` seems to be quite easy.
> it's not possible to reliably test that a program does not perform some action if there is no time boundary (in this case test that task never completes if source thread never completes).
What's the problem with waiting on the latch as I proposed?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5d78f3b6765a029761a3d7abcea9ba71ccd73e6e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806)
* 35c875a260b4dca9cd86a67356fa9488f2ea5aae UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646593243
Thanks for the review @pnowojski, I've addressed your comments (1 unresolved).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
}, {
"hash" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906",
"triggerID" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"triggerType" : "PUSH"
}, {
"hash" : "0d85f9ed9df65e13533442c95cc904ee1093aa16",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0d85f9ed9df65e13533442c95cc904ee1093aa16",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eaf54b89a544a57afc928a689c14e40903c1f73d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906)
* 0d85f9ed9df65e13533442c95cc904ee1093aa16 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r442782518
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -657,20 +657,25 @@ protected void cleanUpInvoke() throws Exception {
}
@Override
- public final void cancel() throws Exception {
+ public final CompletableFuture<Void> cancel() {
isRunning = false;
canceled = true;
// the "cancel task" call must come first, but the cancelables must be
// closed no matter what
- try {
- cancelTask();
- }
- finally {
- mailboxProcessor.allActionsCompleted();
- cancelables.close();
+ return FutureUtils
+ .completedVoidFuture()
+ .thenRun(this::cancelTask)
Review comment:
Because `cancelTask` can also throw an exception.
Leaving client code to deal with both method throwing an exception and Future completed exceptionally isn't very nice IMO.
Having a `finally` block (I guess that's what you mean) would make calling `mailboxProcessor.allActionsCompleted` more complicated (I think it should be called after future completes if it was created successfully).
This is btw an idiomatic way in some languages like Scala.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3833",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
}, {
"hash" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906",
"triggerID" : "eaf54b89a544a57afc928a689c14e40903c1f73d",
"triggerType" : "PUSH"
}, {
"hash" : "0d85f9ed9df65e13533442c95cc904ee1093aa16",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3965",
"triggerID" : "0d85f9ed9df65e13533442c95cc904ee1093aa16",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eaf54b89a544a57afc928a689c14e40903c1f73d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3906)
* 0d85f9ed9df65e13533442c95cc904ee1093aa16 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3965)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r442817643
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -819,6 +823,8 @@ else if (transitionState(current, ExecutionState.FAILED, t)) {
}
finally {
try {
+ waitInvokableCancelCompletion();
Review comment:
This case is already covered by `TaskTest.testTerminationFutureCompletesOnNormalExecution`.
I'll add a bit waiting and re-check there.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12714:
URL: https://github.com/apache/flink/pull/12714#issuecomment-646493259
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806",
"triggerID" : "5d78f3b6765a029761a3d7abcea9ba71ccd73e6e",
"triggerType" : "PUSH"
}, {
"hash" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825",
"triggerID" : "35c875a260b4dca9cd86a67356fa9488f2ea5aae",
"triggerType" : "PUSH"
}, {
"hash" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5d78f3b6765a029761a3d7abcea9ba71ccd73e6e Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3806)
* 35c875a260b4dca9cd86a67356fa9488f2ea5aae Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3825)
* 2a05ea57a26130ae3c4f46cf464dc16f2a9ae84f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #12714: [FLINK-15467][task] Wait for invokable cancellation when stopping Task
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #12714:
URL: https://github.com/apache/flink/pull/12714#discussion_r443462651
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
##########
@@ -871,6 +901,9 @@ public void testTerminationFutureCompletesOnNormalExecution() throws Exception {
assertFalse(task.getTerminationFuture().isDone());
+ task.getExecutingThread().join(500); // allow some arbitrary delay for potential termination and re-check
+ assertFalse(task.getTerminationFuture().isDone());
Review comment:
Removed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org