You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2021/07/13 21:52:16 UTC
[flink] branch master updated (b1c5421 -> 6451a1c)
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from b1c5421 [hotfix][runtime] Log slot pool status if unable to fulfill job requirements
new 4b867d8 [FLINK-22545][coordination] Fix check during creation of Source Coordinator thread.
new 6451a1c [hotfix] Remove unnecessary warning suppression.
The 2 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:
.../flink/util/FatalExitExceptionHandler.java | 1 -
.../coordinator/SourceCoordinatorProvider.java | 33 +++++++++++++++++-----
2 files changed, 26 insertions(+), 8 deletions(-)
[flink] 01/02: [FLINK-22545][coordination] Fix check during
creation of Source Coordinator thread.
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4b867d8a6c187d6bef45294a6e6a524254410167
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Tue Jul 13 14:36:32 2021 +0200
[FLINK-22545][coordination] Fix check during creation of Source Coordinator thread.
The check was meant as a safeguard to prevent re-instantiation after fatal errors killed a previous thread.
But the check was susceptible to thread termination due to idleness in the executor.
This updates the check to only fail if there is in fact an instantiation next to a running thread, or after a
previously crashed thread.
---
.../coordinator/SourceCoordinatorProvider.java | 33 +++++++++++++++++-----
1 file changed, 26 insertions(+), 7 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
index bb6f835..56248f6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorProvider.java
@@ -27,6 +27,8 @@ import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
import org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator;
import org.apache.flink.util.FatalExitExceptionHandler;
+import javax.annotation.Nullable;
+
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -85,13 +87,16 @@ public class SourceCoordinatorProvider<SplitT extends SourceSplit>
}
/** A thread factory class that provides some helper methods. */
- public static class CoordinatorExecutorThreadFactory implements ThreadFactory {
+ public static class CoordinatorExecutorThreadFactory
+ implements ThreadFactory, Thread.UncaughtExceptionHandler {
private final String coordinatorThreadName;
private final ClassLoader cl;
private final Thread.UncaughtExceptionHandler errorHandler;
- private Thread t;
+ @Nullable private Thread t;
+
+ @Nullable private volatile Throwable previousFailureReason;
CoordinatorExecutorThreadFactory(
final String coordinatorThreadName, final ClassLoader contextClassLoader) {
@@ -110,18 +115,32 @@ public class SourceCoordinatorProvider<SplitT extends SourceSplit>
@Override
public synchronized Thread newThread(Runnable r) {
- if (t != null) {
+ if (t != null && t.isAlive()) {
+ throw new Error(
+ "Source Coordinator Thread already exists. There should never be more than one "
+ + "thread driving the actions of a Source Coordinator. Existing Thread: "
+ + t);
+ }
+ if (t != null && previousFailureReason != null) {
throw new Error(
- "This indicates that a fatal error has happened and caused the "
- + "coordinator executor thread to exit. Check the earlier logs "
- + "to see the root cause of the problem.");
+ "The following fatal error has happened in a previously spawned "
+ + "Source Coordinator thread. No new thread can be spawned.",
+ previousFailureReason);
}
t = new Thread(r, coordinatorThreadName);
t.setContextClassLoader(cl);
- t.setUncaughtExceptionHandler(errorHandler);
+ t.setUncaughtExceptionHandler(this);
return t;
}
+ @Override
+ public synchronized void uncaughtException(Thread t, Throwable e) {
+ if (previousFailureReason == null) {
+ previousFailureReason = e;
+ }
+ errorHandler.uncaughtException(t, e);
+ }
+
String getCoordinatorThreadName() {
return coordinatorThreadName;
}
[flink] 02/02: [hotfix] Remove unnecessary warning suppression.
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6451a1c169fbde44c2c7c458cc717c0fcad01cf9
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Tue Jul 13 15:05:25 2021 +0200
[hotfix] Remove unnecessary warning suppression.
This suppression was added to supress compiler warning about 'finally' not returning from 'System.exit()'.
With the introduction of the indirection to the security manager, this compiler warning no longer happens.
---
.../src/main/java/org/apache/flink/util/FatalExitExceptionHandler.java | 1 -
1 file changed, 1 deletion(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/FatalExitExceptionHandler.java b/flink-core/src/main/java/org/apache/flink/util/FatalExitExceptionHandler.java
index 2433398..8ff0e9b 100644
--- a/flink-core/src/main/java/org/apache/flink/util/FatalExitExceptionHandler.java
+++ b/flink-core/src/main/java/org/apache/flink/util/FatalExitExceptionHandler.java
@@ -37,7 +37,6 @@ public final class FatalExitExceptionHandler implements Thread.UncaughtException
public static final int EXIT_CODE = -17;
@Override
- @SuppressWarnings("finally")
public void uncaughtException(Thread t, Throwable e) {
try {
LOG.error(