You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2021/04/26 09:49:38 UTC
[flink] branch release-1.13 updated (3523f5e -> a070dfa)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 3523f5e [hotfix] Do not use ExecutorService.submit since it can swallow exceptions
new 72fc287 [FLINK-22431] Add information when and why the AdaptiveScheduler restarts or fails jobs
new 3795a71 [hotfix] Add debug logging to the states of the AdaptiveScheduler
new a070dfa [hotfix] Harden against FLINK-21376 by checking for null failure cause
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../scheduler/adaptive/AdaptiveScheduler.java | 18 +++++++++++++----
.../runtime/scheduler/adaptive/Canceling.java | 6 +++++-
.../runtime/scheduler/adaptive/Executing.java | 23 +++++++++++++---------
.../flink/runtime/scheduler/adaptive/Failing.java | 6 +++++-
.../flink/runtime/scheduler/adaptive/Finished.java | 7 ++++++-
.../runtime/scheduler/adaptive/Restarting.java | 6 +++++-
.../adaptive/StateWithExecutionGraph.java | 4 ++++
.../scheduler/adaptive/StopWithSavepoint.java | 8 +++++++-
.../runtime/scheduler/adaptive/ExecutingTest.java | 5 +++--
.../scheduler/adaptive/StopWithSavepointTest.java | 6 +++---
10 files changed, 66 insertions(+), 23 deletions(-)
[flink] 02/03: [hotfix] Add debug logging to the states of the
AdaptiveScheduler
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3795a71680be939912101361f2d3bc192b973991
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Fri Apr 23 16:12:17 2021 +0200
[hotfix] Add debug logging to the states of the AdaptiveScheduler
This commit adds debug log statements to the states of the AdaptiveScheduler to log
whenever we ignore a global failure.
---
.../runtime/scheduler/adaptive/AdaptiveScheduler.java | 16 +++++++++++++---
.../flink/runtime/scheduler/adaptive/Canceling.java | 6 +++++-
.../apache/flink/runtime/scheduler/adaptive/Failing.java | 6 +++++-
.../flink/runtime/scheduler/adaptive/Finished.java | 7 ++++++-
.../flink/runtime/scheduler/adaptive/Restarting.java | 6 +++++-
.../scheduler/adaptive/StateWithExecutionGraph.java | 4 ++++
6 files changed, 38 insertions(+), 7 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index c7da630..9a1a252 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -1035,14 +1035,24 @@ public class AdaptiveScheduler
@Override
public void onFinished(ArchivedExecutionGraph archivedExecutionGraph) {
+
+ @Nullable
+ final Throwable optionalFailure =
+ archivedExecutionGraph.getFailureInfo() != null
+ ? archivedExecutionGraph.getFailureInfo().getException()
+ : null;
+ LOG.info(
+ "Job {} reached terminal state {}.",
+ archivedExecutionGraph.getJobID(),
+ archivedExecutionGraph.getState(),
+ optionalFailure);
+
if (jobStatusListener != null) {
jobStatusListener.jobStatusChanges(
jobInformation.getJobID(),
archivedExecutionGraph.getState(),
archivedExecutionGraph.getStatusTimestamp(archivedExecutionGraph.getState()),
- archivedExecutionGraph.getFailureInfo() != null
- ? archivedExecutionGraph.getFailureInfo().getException()
- : null);
+ optionalFailure);
}
jobTerminationFuture.complete(archivedExecutionGraph.getState());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Canceling.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Canceling.java
index 54ddabc..f5d82a1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Canceling.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Canceling.java
@@ -56,7 +56,11 @@ class Canceling extends StateWithExecutionGraph {
@Override
public void handleGlobalFailure(Throwable cause) {
- // ignore global failures
+ getLogger()
+ .debug(
+ "Ignored global failure because we are already canceling the job {}.",
+ getJobId(),
+ cause);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Failing.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Failing.java
index dc44c3c..c836161 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Failing.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Failing.java
@@ -58,7 +58,11 @@ class Failing extends StateWithExecutionGraph {
@Override
public void handleGlobalFailure(Throwable cause) {
- // nothing to do since we are already failing
+ getLogger()
+ .debug(
+ "Ignored global failure because we are already failing the job {}.",
+ getJobId(),
+ cause);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java
index 7a7c9f2..f36a98b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java
@@ -54,7 +54,12 @@ class Finished implements State {
}
@Override
- public void handleGlobalFailure(Throwable cause) {}
+ public void handleGlobalFailure(Throwable cause) {
+ logger.debug(
+ "Ignore global failure because we already finished the job {}.",
+ archivedExecutionGraph.getJobID(),
+ cause);
+ }
@Override
public Logger getLogger() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java
index ffd7eb6..ed5146a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java
@@ -77,7 +77,11 @@ class Restarting extends StateWithExecutionGraph {
@Override
public void handleGlobalFailure(Throwable cause) {
- // don't do anything
+ getLogger()
+ .debug(
+ "Ignored global failure because we are already restarting the job {}.",
+ getJobId(),
+ cause);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
index a85caac..9962c78 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
@@ -112,6 +112,10 @@ abstract class StateWithExecutionGraph implements State {
return executionGraph;
}
+ JobID getJobId() {
+ return executionGraph.getJobID();
+ }
+
protected OperatorCoordinatorHandler getOperatorCoordinatorHandler() {
return operatorCoordinatorHandler;
}
[flink] 01/03: [FLINK-22431] Add information when and why the
AdaptiveScheduler restarts or fails jobs
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 72fc287e06a3bc124d850df9c53bdea3b3c13393
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Fri Apr 23 16:00:56 2021 +0200
[FLINK-22431] Add information when and why the AdaptiveScheduler restarts or fails jobs
This commit adds info log statements to tell the user when and why it restarts or fails a job.
This closes #15736.
---
.../runtime/scheduler/adaptive/AdaptiveScheduler.java | 2 +-
.../flink/runtime/scheduler/adaptive/Executing.java | 16 ++++++++--------
.../runtime/scheduler/adaptive/StopWithSavepoint.java | 2 ++
.../flink/runtime/scheduler/adaptive/ExecutingTest.java | 5 +++--
.../scheduler/adaptive/StopWithSavepointTest.java | 6 +++---
5 files changed, 17 insertions(+), 14 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index bad1cc9..c7da630 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -1058,7 +1058,7 @@ public class AdaptiveScheduler
restartBackoffTimeStrategy.notifyFailure(failure);
if (restartBackoffTimeStrategy.canRestart()) {
return Executing.FailureResult.canRestart(
- Duration.ofMillis(restartBackoffTimeStrategy.getBackoffTime()));
+ failure, Duration.ofMillis(restartBackoffTimeStrategy.getBackoffTime()));
} else {
return Executing.FailureResult.canNotRestart(
new JobException(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
index dcb0366..83beda3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
@@ -85,12 +85,14 @@ class Executing extends StateWithExecutionGraph implements ResourceConsumer {
final FailureResult failureResult = context.howToHandleFailure(cause);
if (failureResult.canRestart()) {
+ getLogger().info("Restarting job.", failureResult.getFailureCause());
context.goToRestarting(
getExecutionGraph(),
getExecutionGraphHandler(),
getOperatorCoordinatorHandler(),
failureResult.getBackoffTime());
} else {
+ getLogger().info("Failing job.", failureResult.getFailureCause());
context.goToFailing(
getExecutionGraph(),
getExecutionGraphHandler(),
@@ -281,9 +283,9 @@ class Executing extends StateWithExecutionGraph implements ResourceConsumer {
static final class FailureResult {
@Nullable private final Duration backoffTime;
- @Nullable private final Throwable failureCause;
+ private final Throwable failureCause;
- private FailureResult(@Nullable Duration backoffTime, @Nullable Throwable failureCause) {
+ private FailureResult(Throwable failureCause, @Nullable Duration backoffTime) {
this.backoffTime = backoffTime;
this.failureCause = failureCause;
}
@@ -299,20 +301,18 @@ class Executing extends StateWithExecutionGraph implements ResourceConsumer {
}
Throwable getFailureCause() {
- Preconditions.checkState(
- failureCause != null,
- "Failure result must not be restartable to return a failure cause.");
return failureCause;
}
/**
* Creates a FailureResult which allows to restart the job.
*
+ * @param failureCause failureCause for restarting the job
* @param backoffTime backoffTime to wait before restarting the job
* @return FailureResult which allows to restart the job
*/
- static FailureResult canRestart(Duration backoffTime) {
- return new FailureResult(backoffTime, null);
+ static FailureResult canRestart(Throwable failureCause, Duration backoffTime) {
+ return new FailureResult(failureCause, backoffTime);
}
/**
@@ -322,7 +322,7 @@ class Executing extends StateWithExecutionGraph implements ResourceConsumer {
* @return FailureResult which does not allow to restart the job
*/
static FailureResult canNotRestart(Throwable failureCause) {
- return new FailureResult(null, failureCause);
+ return new FailureResult(failureCause, null);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java
index dcde71c..ab4f937 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java
@@ -179,12 +179,14 @@ class StopWithSavepoint extends StateWithExecutionGraph {
final Executing.FailureResult failureResult = context.howToHandleFailure(cause);
if (failureResult.canRestart()) {
+ getLogger().info("Restarting job.", failureResult.getFailureCause());
context.goToRestarting(
getExecutionGraph(),
getExecutionGraphHandler(),
getOperatorCoordinatorHandler(),
failureResult.getBackoffTime());
} else {
+ getLogger().info("Failing job.", failureResult.getFailureCause());
context.goToFailing(
getExecutionGraph(),
getExecutionGraphHandler(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
index df5f233..2852bda 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
@@ -140,7 +140,7 @@ public class ExecutingTest extends TestLogger {
ctx.setExpectRestarting(
(restartingArguments ->
assertThat(restartingArguments.getBackoffTime(), is(duration))));
- ctx.setHowToHandleFailure((t) -> Executing.FailureResult.canRestart(duration));
+ ctx.setHowToHandleFailure((t) -> Executing.FailureResult.canRestart(t, duration));
exec.handleGlobalFailure(new RuntimeException("Recoverable error"));
}
}
@@ -234,7 +234,8 @@ public class ExecutingTest extends TestLogger {
new ExecutingStateBuilder()
.setExecutionGraph(returnsFailedStateExecutionGraph)
.build(ctx);
- ctx.setHowToHandleFailure((ign) -> Executing.FailureResult.canRestart(Duration.ZERO));
+ ctx.setHowToHandleFailure(
+ (throwable) -> Executing.FailureResult.canRestart(throwable, Duration.ZERO));
ctx.setExpectRestarting(assertNonNull());
exec.updateTaskExecutionState(createFailingStateTransition());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepointTest.java
index e0383fa..e9a1157 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepointTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepointTest.java
@@ -174,7 +174,7 @@ public class StopWithSavepointTest extends TestLogger {
StopWithSavepoint sws = createStopWithSavepoint(ctx);
ctx.setStopWithSavepoint(sws);
ctx.setHowToHandleFailure(
- (ignore) -> Executing.FailureResult.canRestart(Duration.ZERO));
+ (throwable) -> Executing.FailureResult.canRestart(throwable, Duration.ZERO));
ctx.setExpectRestarting(assertNonNull());
@@ -229,7 +229,7 @@ public class StopWithSavepointTest extends TestLogger {
createStopWithSavepoint(ctx, new StateTrackingMockExecutionGraph());
ctx.setStopWithSavepoint(sws);
ctx.setHowToHandleFailure(
- (ignore) -> Executing.FailureResult.canRestart(Duration.ZERO));
+ (throwable) -> Executing.FailureResult.canRestart(throwable, Duration.ZERO));
ctx.setExpectRestarting(assertNonNull());
@@ -277,7 +277,7 @@ public class StopWithSavepointTest extends TestLogger {
ctx.setStopWithSavepoint(sws);
ctx.setHowToHandleFailure(
- (ignore) -> Executing.FailureResult.canRestart(Duration.ZERO));
+ (throwable) -> Executing.FailureResult.canRestart(throwable, Duration.ZERO));
ctx.setExpectRestarting(assertNonNull());
[flink] 03/03: [hotfix] Harden against FLINK-21376 by checking for
null failure cause
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a070dfaae6ae063daab168557a66d827aa0aa61d
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Fri Apr 23 17:50:15 2021 +0200
[hotfix] Harden against FLINK-21376 by checking for null failure cause
In order to harden the AdaptiveScheduler against FLINK-21376, this commit checks whether a task
failure cause is null or not. In case of null, it will replace the failure with a generic cause.
---
.../org/apache/flink/runtime/scheduler/adaptive/Executing.java | 7 ++++++-
.../apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java | 6 +++++-
2 files changed, 11 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
index 83beda3..6b44c51 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
import org.apache.flink.runtime.scheduler.ExecutionGraphHandler;
import org.apache.flink.runtime.scheduler.OperatorCoordinatorHandler;
import org.apache.flink.runtime.scheduler.stopwithsavepoint.StopWithSavepointTerminationManager;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -108,7 +109,11 @@ class Executing extends StateWithExecutionGraph implements ResourceConsumer {
if (successfulUpdate) {
if (taskExecutionState.getExecutionState() == ExecutionState.FAILED) {
Throwable cause = taskExecutionState.getError(userCodeClassLoader);
- handleAnyFailure(cause);
+ handleAnyFailure(
+ cause == null
+ ? new FlinkException(
+ "Unknown failure cause. Probably related to FLINK-21376.")
+ : cause);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java
index ab4f937..6c8dc4f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java
@@ -147,7 +147,11 @@ class StopWithSavepoint extends StateWithExecutionGraph {
if (successfulUpdate) {
if (taskExecutionStateTransition.getExecutionState() == ExecutionState.FAILED) {
Throwable cause = taskExecutionStateTransition.getError(userCodeClassLoader);
- handleAnyFailure(cause);
+ handleAnyFailure(
+ cause == null
+ ? new FlinkException(
+ "Unknown failure cause. Probably related to FLINK-21376.")
+ : cause);
}
}