You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/10/08 08:01:22 UTC
[flink] 08/10: [FLINK-24182][task] Unify failureCause handling and
improve logging
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a2233912bc33a092705813a34a56be90db0f8008
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Sep 15 06:42:43 2021 +0200
[FLINK-24182][task] Unify failureCause handling and improve logging
Now failureCause is set in only one place and catching CancelTaskException
gives more detailed logs.
---
flink-end-to-end-tests/test-scripts/common.sh | 1 +
.../org/apache/flink/runtime/taskmanager/Task.java | 29 +++++++++++++++-------
2 files changed, 21 insertions(+), 9 deletions(-)
diff --git a/flink-end-to-end-tests/test-scripts/common.sh b/flink-end-to-end-tests/test-scripts/common.sh
index 30f9e36..4dfb311 100644
--- a/flink-end-to-end-tests/test-scripts/common.sh
+++ b/flink-end-to-end-tests/test-scripts/common.sh
@@ -382,6 +382,7 @@ function check_logs_for_errors {
function check_logs_for_exceptions {
echo "Checking for exceptions..."
exception_count=$(grep -rv "GroupCoordinatorNotAvailableException" $FLINK_LOG_DIR \
+ | grep -v "due to CancelTaskException" \
| grep -v "RetriableCommitFailedException" \
| grep -v "NoAvailableBrokersException" \
| grep -v "Async Kafka commit failed" \
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 13ee5d7..e28c71f 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
@@ -796,18 +796,15 @@ public class Task
if (current == ExecutionState.RUNNING
|| current == ExecutionState.INITIALIZING
|| current == ExecutionState.DEPLOYING) {
- if (t instanceof CancelTaskException) {
- if (transitionState(current, ExecutionState.CANCELED)) {
+ if (ExceptionUtils.findThrowable(t, CancelTaskException.class)
+ .isPresent()) {
+ if (transitionState(current, ExecutionState.CANCELED, t)) {
cancelInvokable(invokable);
break;
}
} else {
if (transitionState(current, ExecutionState.FAILED, t)) {
- // proper failure of the task. record the exception as the root
- // cause
- failureCause = t;
cancelInvokable(invokable);
-
break;
}
}
@@ -1101,7 +1098,24 @@ public class Task
executionId,
currentState,
newState);
+ } else if (ExceptionUtils.findThrowable(cause, CancelTaskException.class).isPresent()) {
+ LOG.info(
+ "{} ({}) switched from {} to {} due to CancelTaskException.",
+ taskNameWithSubtask,
+ executionId,
+ currentState,
+ newState);
+ LOG.debug(
+ "{} ({}) switched from {} to {} due to CancelTaskException: {}",
+ taskNameWithSubtask,
+ executionId,
+ currentState,
+ newState,
+ ExceptionUtils.stringifyException(cause));
} else {
+ // proper failure of the task. record the exception as the root
+ // cause
+ failureCause = cause;
LOG.warn(
"{} ({}) switched from {} to {} with failure cause: {}",
taskNameWithSubtask,
@@ -1183,7 +1197,6 @@ public class Task
if (transitionState(current, targetState, cause)) {
// if we manage this state transition, then the invokable gets never called
// we need not call cancel on it
- this.failureCause = cause;
return;
}
} else if (current == ExecutionState.INITIALIZING
@@ -1196,8 +1209,6 @@ public class Task
final TaskInvokable invokable = this.invokable;
if (invokable != null && invokableHasBeenCanceled.compareAndSet(false, true)) {
- this.failureCause = cause;
-
LOG.info(
"Triggering cancellation of task code {} ({}).",
taskNameWithSubtask,