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 2020/05/06 07:24:12 UTC
[flink] branch release-1.10 updated (99ea1aa -> 48ae089)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 99ea1aa [FLINK-17496][kinesis] Upgrade amazon-kinesis-producer to 0.14.0
new 2e42cfa [hotfix][tests] Add TestingTaskManagerActions implementation
new bd76b42 [FLINK-17514] Let TaskCancelerWatchDog call TaskManagerActions.notifyFatalError with non-null cause
new f3f1bd3 [FLINK-17514] Fail fatally if the TaskCancelerWatchDog encounters exception in run method
new 48ae089 [FLINK-17514] Harden ExceptionUtils.tryEnrichTaskManagerError, .isMetaspaceOutOfMemoryError and .isDirectOutOfMemoryError to handle null values
The 4 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/core/memory/MemorySegmentFactory.java | 7 +-
.../java/org/apache/flink/util/ExceptionUtils.java | 50 +++++++-----
.../org/apache/flink/util/ExceptionUtilsTest.java | 16 ++++
.../runtime/taskexecutor/TaskManagerRunner.java | 2 +-
.../org/apache/flink/runtime/taskmanager/Task.java | 19 ++---
.../apache/flink/runtime/taskmanager/TaskTest.java | 40 ++++------
.../taskmanager/TestingTaskManagerActions.java | 92 ++++++++++++++++++++++
7 files changed, 168 insertions(+), 58 deletions(-)
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestingTaskManagerActions.java
[flink] 02/04: [FLINK-17514] Let TaskCancelerWatchDog call
TaskManagerActions.notifyFatalError with non-null 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.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bd76b42228bde4f25969d65712f7799771fb91e9
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue May 5 11:44:41 2020 +0200
[FLINK-17514] Let TaskCancelerWatchDog call TaskManagerActions.notifyFatalError with non-null cause
In order to avoid NPE we have to call TaskManagerActions.notifyFatalError with non-null arguments. This
commit changes the behaviour accordingly.
---
.../org/apache/flink/runtime/taskmanager/Task.java | 4 +--
.../apache/flink/runtime/taskmanager/TaskTest.java | 40 +++++++++-------------
2 files changed, 19 insertions(+), 25 deletions(-)
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 689ada8..0b35df0 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
@@ -75,6 +75,7 @@ import org.apache.flink.runtime.util.FatalExitExceptionHandler;
import org.apache.flink.types.Either;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
import org.apache.flink.util.WrappingRuntimeException;
@@ -1504,8 +1505,7 @@ public class Task implements Runnable, TaskSlotPayload, TaskActions, PartitionPr
if (executerThread.isAlive()) {
String msg = "Task did not exit gracefully within " + (timeoutMillis / 1000) + " + seconds.";
- log.error(msg);
- taskManager.notifyFatalError(msg, null);
+ taskManager.notifyFatalError(msg, new FlinkRuntimeException(msg));
}
}
catch (Throwable t) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
index 41ae741..3ca2caa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
@@ -75,7 +75,9 @@ import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -798,12 +800,13 @@ public class TaskTest extends TestLogger {
*/
@Test
public void testFatalErrorAfterUnInterruptibleInvoke() throws Exception {
- final AwaitFatalErrorTaskManagerActions taskManagerActions =
- new AwaitFatalErrorTaskManagerActions();
+ final CompletableFuture<Throwable> fatalErrorFuture = new CompletableFuture<>();
+ final TestingTaskManagerActions taskManagerActions = TestingTaskManagerActions.newBuilder()
+ .setNotifyFatalErrorConsumer((s, throwable) -> fatalErrorFuture.complete(throwable))
+ .build();
final Configuration config = new Configuration();
- config.setLong(TaskManagerOptions.TASK_CANCELLATION_INTERVAL, 5);
- config.setLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT, 50);
+ config.setLong(TaskManagerOptions.TASK_CANCELLATION_TIMEOUT, 10);
final Task task = createTaskBuilder()
.setInvokable(InvokableUnInterruptibleBlockingInvoke.class)
@@ -819,7 +822,8 @@ public class TaskTest extends TestLogger {
task.cancelExecution();
// wait for the notification of notifyFatalError
- taskManagerActions.latch.await();
+ final Throwable fatalError = fatalErrorFuture.join();
+ assertThat(fatalError, is(notNullValue()));
} finally {
// Interrupt again to clean up Thread
triggerLatch.trigger();
@@ -833,8 +837,10 @@ public class TaskTest extends TestLogger {
*/
@Test
public void testFatalErrorOnCanceling() throws Exception {
- final AwaitFatalErrorTaskManagerActions taskManagerActions =
- new AwaitFatalErrorTaskManagerActions();
+ final CompletableFuture<Throwable> fatalErrorFuture = new CompletableFuture<>();
+ final TestingTaskManagerActions taskManagerActions = TestingTaskManagerActions.newBuilder()
+ .setNotifyFatalErrorConsumer((s, throwable) -> fatalErrorFuture.complete(throwable))
+ .build();
final Configuration config = new Configuration();
config.setLong(TaskManagerOptions.TASK_CANCELLATION_INTERVAL, 5);
@@ -846,7 +852,8 @@ public class TaskTest extends TestLogger {
.setTaskManagerActions(taskManagerActions)
.build());
- doThrow(OutOfMemoryError.class).when(task).cancelOrFailAndCancelInvokableInternal(eq(ExecutionState.CANCELING), eq(null));
+ final Class<OutOfMemoryError> fatalErrorType = OutOfMemoryError.class;
+ doThrow(fatalErrorType).when(task).cancelOrFailAndCancelInvokableInternal(eq(ExecutionState.CANCELING), eq(null));
try {
task.startTaskThread();
@@ -856,9 +863,8 @@ public class TaskTest extends TestLogger {
task.cancelExecution();
// wait for the notification of notifyFatalError
- taskManagerActions.latch.await();
- } catch (Throwable t) {
- fail("No exception is expected to be thrown by fatal error handling");
+ final Throwable fatalError = fatalErrorFuture.join();
+ assertThat(fatalError, instanceOf(fatalErrorType));
} finally {
triggerLatch.trigger();
}
@@ -1010,18 +1016,6 @@ public class TaskTest extends TestLogger {
}
}
- /**
- * Customized TaskManagerActions that waits for a call of notifyFatalError.
- */
- private static class AwaitFatalErrorTaskManagerActions extends NoOpTaskManagerActions {
- private final OneShotLatch latch = new OneShotLatch();
-
- @Override
- public void notifyFatalError(String message, Throwable cause) {
- latch.trigger();
- }
- }
-
// ------------------------------------------------------------------------
// helper functions
// ------------------------------------------------------------------------
[flink] 01/04: [hotfix][tests] Add TestingTaskManagerActions
implementation
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.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2e42cfa78d46c0d227ea2506af715fbd647c1d5f
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue May 5 11:44:07 2020 +0200
[hotfix][tests] Add TestingTaskManagerActions implementation
---
.../taskmanager/TestingTaskManagerActions.java | 92 ++++++++++++++++++++++
1 file changed, 92 insertions(+)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestingTaskManagerActions.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestingTaskManagerActions.java
new file mode 100644
index 0000000..3901430
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestingTaskManagerActions.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskmanager;
+
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+/**
+ * Testing implementation of {@link TaskManagerActions}.
+ */
+public class TestingTaskManagerActions implements TaskManagerActions {
+
+ private final BiConsumer<String, Throwable> notifyFatalErrorConsumer;
+
+ private final BiConsumer<ExecutionAttemptID, Throwable> failTaskConsumer;
+
+ private final Consumer<TaskExecutionState> updateTaskExecutionStateConsumer;
+
+ private TestingTaskManagerActions(
+ BiConsumer<String, Throwable> notifyFatalErrorConsumer,
+ BiConsumer<ExecutionAttemptID, Throwable> failTaskConsumer,
+ Consumer<TaskExecutionState> updateTaskExecutionStateConsumer) {
+ this.notifyFatalErrorConsumer = notifyFatalErrorConsumer;
+ this.failTaskConsumer = failTaskConsumer;
+ this.updateTaskExecutionStateConsumer = updateTaskExecutionStateConsumer;
+ }
+
+ @Override
+ public void notifyFatalError(String message, Throwable cause) {
+ notifyFatalErrorConsumer.accept(message, cause);
+ }
+
+ @Override
+ public void failTask(ExecutionAttemptID executionAttemptID, Throwable cause) {
+ failTaskConsumer.accept(executionAttemptID, cause);
+ }
+
+ @Override
+ public void updateTaskExecutionState(TaskExecutionState taskExecutionState) {
+ updateTaskExecutionStateConsumer.accept(taskExecutionState);
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private BiConsumer<String, Throwable> notifyFatalErrorConsumer = (ignoredA, ignoredB) -> {};
+ private BiConsumer<ExecutionAttemptID, Throwable> failTaskConsumer = (ignoredA, ignoredB) -> {};
+ private Consumer<TaskExecutionState> updateTaskExecutionStateConsumer = ignored -> {};
+
+ private Builder() {}
+
+ public Builder setNotifyFatalErrorConsumer(BiConsumer<String, Throwable> notifyFatalErrorConsumer) {
+ this.notifyFatalErrorConsumer = notifyFatalErrorConsumer;
+ return this;
+ }
+
+ public Builder setFailTaskConsumer(BiConsumer<ExecutionAttemptID, Throwable> failTaskConsumer) {
+ this.failTaskConsumer = failTaskConsumer;
+ return this;
+ }
+
+ public Builder setUpdateTaskExecutionStateConsumer(Consumer<TaskExecutionState> updateTaskExecutionStateConsumer) {
+ this.updateTaskExecutionStateConsumer = updateTaskExecutionStateConsumer;
+ return this;
+ }
+
+ public TestingTaskManagerActions build() {
+ return new TestingTaskManagerActions(notifyFatalErrorConsumer, failTaskConsumer, updateTaskExecutionStateConsumer);
+ }
+ }
+}
[flink] 03/04: [FLINK-17514] Fail fatally if the
TaskCancelerWatchDog encounters exception in run method
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.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f3f1bd3020c473d65fe8963262b8962d20e84d6d
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue May 5 11:47:12 2020 +0200
[FLINK-17514] Fail fatally if the TaskCancelerWatchDog encounters exception in run method
If the TaskCancelerWatchDog encounters an exception in the run method, then we can no longer
guarantee that it will do its job. Hence, it is best to fail fatally by letting the exception
bubble up so that it is handled by the uncaught exception handler.
---
.../java/org/apache/flink/runtime/taskmanager/Task.java | 13 +++----------
1 file changed, 3 insertions(+), 10 deletions(-)
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 0b35df0..d7f3d65 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
@@ -1082,8 +1082,7 @@ public class Task implements Runnable, TaskSlotPayload, TaskActions, PartitionPr
Runnable cancelWatchdog = new TaskCancelerWatchDog(
executingThread,
taskManagerActions,
- taskCancellationTimeout,
- LOG);
+ taskCancellationTimeout);
Thread watchDogThread = new Thread(
executingThread.getThreadGroup(),
@@ -1460,9 +1459,6 @@ public class Task implements Runnable, TaskSlotPayload, TaskActions, PartitionPr
*/
private static class TaskCancelerWatchDog implements Runnable {
- /** The logger to report on the fatal condition. */
- private final Logger log;
-
/** The executing task thread that we wait for to terminate. */
private final Thread executerThread;
@@ -1475,12 +1471,10 @@ public class Task implements Runnable, TaskSlotPayload, TaskActions, PartitionPr
TaskCancelerWatchDog(
Thread executerThread,
TaskManagerActions taskManager,
- long timeoutMillis,
- Logger log) {
+ long timeoutMillis) {
checkArgument(timeoutMillis > 0);
- this.log = log;
this.executerThread = executerThread;
this.taskManager = taskManager;
this.timeoutMillis = timeoutMillis;
@@ -1509,8 +1503,7 @@ public class Task implements Runnable, TaskSlotPayload, TaskActions, PartitionPr
}
}
catch (Throwable t) {
- ExceptionUtils.rethrowIfFatalError(t);
- log.error("Error in Task Cancellation Watch Dog", t);
+ throw new FlinkRuntimeException("Error in Task Cancellation Watch Dog", t);
}
}
}
[flink] 04/04: [FLINK-17514] Harden
ExceptionUtils.tryEnrichTaskManagerError,
.isMetaspaceOutOfMemoryError and .isDirectOutOfMemoryError to handle null
values
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.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 48ae089e15e9a324897869e9cf2577587aceffbf
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue May 5 12:08:01 2020 +0200
[FLINK-17514] Harden ExceptionUtils.tryEnrichTaskManagerError, .isMetaspaceOutOfMemoryError and .isDirectOutOfMemoryError to handle null values
Error handling code should not fail. Hence, this commit changes the methods ExceptionUtils.tryEnrichTaskManagerError,
ExceptionUtils.isMetaspaceOutOfMemoryError and ExceptionUtils.isDirectOutOfMemoryError to properly handle a null argument.
This closes #11995.
---
.../flink/core/memory/MemorySegmentFactory.java | 7 +--
.../java/org/apache/flink/util/ExceptionUtils.java | 50 ++++++++++++++--------
.../org/apache/flink/util/ExceptionUtilsTest.java | 16 +++++++
.../runtime/taskexecutor/TaskManagerRunner.java | 2 +-
.../org/apache/flink/runtime/taskmanager/Task.java | 2 +-
5 files changed, 54 insertions(+), 23 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
index 760d2ac..2751d9c 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
@@ -112,12 +112,13 @@ public final class MemorySegmentFactory {
// once we find a common way to handle OOM errors in netty threads.
// Here we enrich it to propagate better OOM message to the receiver
// if it happens in a netty thread.
- OutOfMemoryError enrichedOutOfMemoryError = (OutOfMemoryError) ExceptionUtils
- .enrichTaskManagerOutOfMemoryError(outOfMemoryError);
+ Throwable enrichedOutOfMemoryError = ExceptionUtils.tryEnrichTaskManagerError(outOfMemoryError);
if (ExceptionUtils.isDirectOutOfMemoryError(outOfMemoryError)) {
LOG.error("Cannot allocate direct memory segment", enrichedOutOfMemoryError);
}
- throw enrichedOutOfMemoryError;
+
+ ExceptionUtils.rethrow(enrichedOutOfMemoryError);
+ return null;
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java b/flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java
index 0a63ae5..6512531 100644
--- a/flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java
@@ -132,31 +132,41 @@ public final class ExceptionUtils {
}
/**
- * Generates new {@link OutOfMemoryError} with more detailed message.
+ * Tries to enrich the passed exception with additional information.
*
* <p>This method improves error message for direct and metaspace {@link OutOfMemoryError}.
* It adds description of possible causes and ways of resolution.
*
- * @param exception The exception to enrich.
- * @return either enriched exception if needed or the original one.
+ * @param exception exception to enrich if not {@code null}
+ * @return the enriched exception or the original if no additional information could be added;
+ * {@code null} if the argument was {@code null}
*/
- public static Throwable enrichTaskManagerOutOfMemoryError(Throwable exception) {
- if (isMetaspaceOutOfMemoryError(exception)) {
- return changeOutOfMemoryErrorMessage(exception, TM_METASPACE_OOM_ERROR_MESSAGE);
- } else if (isDirectOutOfMemoryError(exception)) {
- return changeOutOfMemoryErrorMessage(exception, TM_DIRECT_OOM_ERROR_MESSAGE);
+ @Nullable
+ public static Throwable tryEnrichTaskManagerError(@Nullable Throwable exception) {
+ if (exception instanceof OutOfMemoryError) {
+ return tryEnrichTaskManagerOutOfMemoryError((OutOfMemoryError) exception);
}
+
return exception;
}
- private static OutOfMemoryError changeOutOfMemoryErrorMessage(Throwable exception, String newMessage) {
- Preconditions.checkArgument(exception instanceof OutOfMemoryError);
- if (exception.getMessage().equals(newMessage)) {
- return (OutOfMemoryError) exception;
+ private static Throwable tryEnrichTaskManagerOutOfMemoryError(OutOfMemoryError oom) {
+ if (isMetaspaceOutOfMemoryError(oom)) {
+ return changeOutOfMemoryErrorMessage(oom, TM_METASPACE_OOM_ERROR_MESSAGE);
+ } else if (isDirectOutOfMemoryError(oom)) {
+ return changeOutOfMemoryErrorMessage(oom, TM_DIRECT_OOM_ERROR_MESSAGE);
+ }
+
+ return oom;
+ }
+
+ private static OutOfMemoryError changeOutOfMemoryErrorMessage(OutOfMemoryError oom, String newMessage) {
+ if (oom.getMessage().equals(newMessage)) {
+ return oom;
}
OutOfMemoryError newError = new OutOfMemoryError(newMessage);
- newError.initCause(exception.getCause());
- newError.setStackTrace(exception.getStackTrace());
+ newError.initCause(oom.getCause());
+ newError.setStackTrace(oom.getStackTrace());
return newError;
}
@@ -166,7 +176,7 @@ public final class ExceptionUtils {
* @param t The exception to check.
* @return True, if the exception is the metaspace {@link OutOfMemoryError}, false otherwise.
*/
- public static boolean isMetaspaceOutOfMemoryError(Throwable t) {
+ public static boolean isMetaspaceOutOfMemoryError(@Nullable Throwable t) {
return isOutOfMemoryErrorWithMessageStartingWith(t, "Metaspace");
}
@@ -176,14 +186,18 @@ public final class ExceptionUtils {
* @param t The exception to check.
* @return True, if the exception is the direct {@link OutOfMemoryError}, false otherwise.
*/
- public static boolean isDirectOutOfMemoryError(Throwable t) {
+ public static boolean isDirectOutOfMemoryError(@Nullable Throwable t) {
return isOutOfMemoryErrorWithMessageStartingWith(t, "Direct buffer memory");
}
- private static boolean isOutOfMemoryErrorWithMessageStartingWith(Throwable t, String prefix) {
+ private static boolean isOutOfMemoryErrorWithMessageStartingWith(@Nullable Throwable t, String prefix) {
// the exact matching of the class is checked to avoid matching any custom subclasses of OutOfMemoryError
// as we are interested in the original exceptions, generated by JVM.
- return t.getClass() == OutOfMemoryError.class && t.getMessage() != null && t.getMessage().startsWith(prefix);
+ return isOutOfMemoryError(t) && t.getMessage() != null && t.getMessage().startsWith(prefix);
+ }
+
+ private static boolean isOutOfMemoryError(@Nullable Throwable t) {
+ return t != null && t.getClass() == OutOfMemoryError.class;
}
/**
diff --git a/flink-core/src/test/java/org/apache/flink/util/ExceptionUtilsTest.java b/flink-core/src/test/java/org/apache/flink/util/ExceptionUtilsTest.java
index fa275d0..fc72fbb 100644
--- a/flink-core/src/test/java/org/apache/flink/util/ExceptionUtilsTest.java
+++ b/flink-core/src/test/java/org/apache/flink/util/ExceptionUtilsTest.java
@@ -22,6 +22,7 @@ import org.junit.Test;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -87,4 +88,19 @@ public class ExceptionUtilsTest extends TestLogger {
assertThat(strippedException, is(equalTo(expectedException)));
}
+
+ @Test
+ public void testTryEnrichTaskExecutorErrorCanHandleNullValue() {
+ assertThat(ExceptionUtils.tryEnrichTaskManagerError(null), is(nullValue()));
+ }
+
+ @Test
+ public void testIsMetaspaceOutOfMemoryErrorCanHandleNullValue() {
+ assertFalse(ExceptionUtils.isMetaspaceOutOfMemoryError(null));
+ }
+
+ @Test
+ public void testIsDirectOutOfMemoryErrorCanHandleNullValue() {
+ assertFalse(ExceptionUtils.isDirectOutOfMemoryError(null));
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index 8ed4fe4..019d186 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -246,7 +246,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
@Override
public void onFatalError(Throwable exception) {
- Throwable enrichedException = ExceptionUtils.enrichTaskManagerOutOfMemoryError(exception);
+ Throwable enrichedException = ExceptionUtils.tryEnrichTaskManagerError(exception);
LOG.error("Fatal error occurred while executing the TaskManager. Shutting it down...", enrichedException);
// In case of the Metaspace OutOfMemoryError, we expect that the graceful shutdown is possible,
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 d7f3d65..609e559 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
@@ -742,7 +742,7 @@ public class Task implements Runnable, TaskSlotPayload, TaskActions, PartitionPr
// an exception was thrown as a side effect of cancelling
// ----------------------------------------------------------------
- t = ExceptionUtils.enrichTaskManagerOutOfMemoryError(t);
+ t = ExceptionUtils.tryEnrichTaskManagerError(t);
try {
// check if the exception is unrecoverable