You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/06/25 02:16:32 UTC
[flink] 01/02: [hotfix] Extract IgnoreShutdownRejectedExecutionHandler.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0a3a62e30cf0afab0e338a8e7eb58af08aa1434c
Author: Weijie Guo <re...@163.com>
AuthorDate: Wed Jun 21 12:28:06 2023 +0800
[hotfix] Extract IgnoreShutdownRejectedExecutionHandler.
---
.../IgnoreShutdownRejectedExecutionHandler.java | 55 ++++++++++++++++++++++
.../flink/changelog/fs/SchedulerFactory.java | 10 ++--
.../partition/hybrid/HsResultPartitionTest.java | 10 +---
.../tiered/shuffle/TieredResultPartitionTest.java | 10 +---
4 files changed, 62 insertions(+), 23 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/concurrent/IgnoreShutdownRejectedExecutionHandler.java b/flink-core/src/main/java/org/apache/flink/util/concurrent/IgnoreShutdownRejectedExecutionHandler.java
new file mode 100644
index 00000000000..795e70a562b
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/util/concurrent/IgnoreShutdownRejectedExecutionHandler.java
@@ -0,0 +1,55 @@
+/*
+ * 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.util.concurrent;
+
+import org.slf4j.Logger;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * Rejected executions are ignored or logged in debug if the executor is {@link
+ * java.util.concurrent.ThreadPoolExecutor#isShutdown shutdown}. Otherwise, {@link
+ * RejectedExecutionException} is thrown.
+ */
+public class IgnoreShutdownRejectedExecutionHandler implements RejectedExecutionHandler {
+ @Nullable private final Logger logger;
+
+ public IgnoreShutdownRejectedExecutionHandler() {
+ this(null);
+ }
+
+ public IgnoreShutdownRejectedExecutionHandler(@Nullable Logger logger) {
+ this.logger = logger;
+ }
+
+ @Override
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+ if (executor.isShutdown()) {
+ if (logger != null) {
+ logger.debug("Execution is rejected because shutdown is in progress");
+ }
+ } else {
+ throw new RejectedExecutionException();
+ }
+ }
+}
diff --git a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/SchedulerFactory.java b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/SchedulerFactory.java
index 77e867e41a2..30ee3aa9ee5 100644
--- a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/SchedulerFactory.java
+++ b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/SchedulerFactory.java
@@ -17,6 +17,8 @@
package org.apache.flink.changelog.fs;
+import org.apache.flink.util.concurrent.IgnoreShutdownRejectedExecutionHandler;
+
import org.slf4j.Logger;
import java.util.concurrent.RejectedExecutionException;
@@ -50,12 +52,6 @@ class SchedulerFactory {
thread.setUncaughtExceptionHandler(INSTANCE);
return thread;
},
- (ignored, executor) -> {
- if (executor.isShutdown()) {
- log.debug("Execution rejected because shutdown is in progress");
- } else {
- throw new RejectedExecutionException();
- }
- });
+ new IgnoreShutdownRejectedExecutionHandler(log));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartitionTest.java
index f4104e7b869..1c41c8a89c3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartitionTest.java
@@ -46,6 +46,7 @@ import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+import org.apache.flink.util.concurrent.IgnoreShutdownRejectedExecutionHandler;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -62,7 +63,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.Queue;
import java.util.Random;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
@@ -106,13 +106,7 @@ class HsResultPartitionTest {
new ScheduledThreadPoolExecutor(
numThreads,
new ExecutorThreadFactory("test-io-scheduler-thread"),
- (ignored, executor) -> {
- if (executor.isShutdown()) {
- // ignore rejected as shutdown.
- } else {
- throw new RejectedExecutionException();
- }
- });
+ new IgnoreShutdownRejectedExecutionHandler());
}
@AfterEach
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartitionTest.java
index d7c9e6766de..f274e99e85e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartitionTest.java
@@ -38,6 +38,7 @@ import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.Tiere
import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+import org.apache.flink.util.concurrent.IgnoreShutdownRejectedExecutionHandler;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -48,7 +49,6 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -90,13 +90,7 @@ class TieredResultPartitionTest {
new ScheduledThreadPoolExecutor(
NUM_THREADS,
new ExecutorThreadFactory("test-io-scheduler-thread"),
- (ignored, executor) -> {
- if (executor.isShutdown()) {
- // ignore rejected as shutdown.
- } else {
- throw new RejectedExecutionException();
- }
- });
+ new IgnoreShutdownRejectedExecutionHandler());
}
@AfterEach