[
https://issues.apache.org/jira/browse/FLINK-37317?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17946906#comment-17946906
]
Hangxiang Yu commented on FLINK-37317:
--------------------------------------
We could track in the before duplicated one.
> Checkpoint asyncOperationsThreadPool has only one thread running
> ----------------------------------------------------------------
>
> Key: FLINK-37317
> URL: https://issues.apache.org/jira/browse/FLINK-37317
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Checkpointing
> Affects Versions: 1.20.1
> Reporter: Jufang He
> Priority: Major
> Attachments: image-2025-02-13-17-23-11-788.png,
> image-2025-02-13-17-25-47-904.png, image-2025-02-13-17-41-27-027.png
>
>
> org.apache.flink.streaming.runtime.tasks.StreamTask#AsyncOperationsThreadPool
> configuration looks unreasonable, cause in fact will only have one thread to
> run. The concurrent checkpoints asyncCheckpointRunnable execution and close
> can only run serially. The thread pool configuration is as follows:
> !image-2025-02-13-17-23-11-788.png|width=600!
> I simulated the execution of the thread pool locally, and the results were as
> follows:
> {code:java}
> package org.apache.flink.streaming.runtime.tasks;
> import org.apache.flink.api.common.JobID;
> import org.apache.flink.util.FatalExitExceptionHandler;
> import org.apache.flink.util.MdcUtils;
> import org.apache.flink.util.concurrent.ExecutorThreadFactory;
> import java.text.SimpleDateFormat;
> import java.util.Date;
> import java.util.concurrent.ExecutorService;
> import java.util.concurrent.LinkedBlockingQueue;
> import java.util.concurrent.ThreadPoolExecutor;
> import java.util.concurrent.TimeUnit;
> public class Test1 {
> public static void main(String[] args) throws Exception {
> ExecutorService asyncOperationsThreadPool =
> MdcUtils.scopeToJob(
> new JobID(),
> new ThreadPoolExecutor(
> 0,
> 2,
> 60L,
> TimeUnit.SECONDS,
> new LinkedBlockingQueue<>(),
> new ExecutorThreadFactory(
> "AsyncOperations", new
> FatalExitExceptionHandler())));
> for (int i = 0; i < 5; i++) {
> int taskId = i;
> Runnable task =
> () -> {
> System.out.println(getCurrentTime() + " Starting task
> " + taskId + " by thread " + Thread.currentThread().getName());
> try {
> Thread.sleep(10000);
> } catch (InterruptedException e) {
> throw new RuntimeException(e);
> }
> System.out.println(getCurrentTime() + " Task " +
> taskId + " finished by thread " + Thread.currentThread().getName());
> };
> asyncOperationsThreadPool.submit(task);
> }
> Thread.sleep(100000);
> }
> private static String getCurrentTime() {
> Date now = new Date();
> SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd
> HH:mm:ss.SSS");
> return sdf.format(now);
> }
> }
> {code}
> results:
> {code:java}
> 2025-04-24 01:28:41.068 Starting task 0 by thread AsyncOperations-thread-1
> 2025-04-24 01:28:51.089 Task 0 finished by thread AsyncOperations-thread-1
> 2025-04-24 01:28:51.091 Starting task 1 by thread AsyncOperations-thread-1
> 2025-04-24 01:29:01.091 Task 1 finished by thread AsyncOperations-thread-1
> 2025-04-24 01:29:01.093 Starting task 2 by thread AsyncOperations-thread-1
> 2025-04-24 01:29:11.096 Task 2 finished by thread AsyncOperations-thread-1
> 2025-04-24 01:29:11.097 Starting task 3 by thread AsyncOperations-thread-1
> 2025-04-24 01:29:21.097 Task 3 finished by thread AsyncOperations-thread-1
> 2025-04-24 01:29:21.098 Starting task 4 by thread AsyncOperations-thread-1
> 2025-04-24 01:29:31.102 Task 4 finished by thread AsyncOperations-thread-1
> {code}
> I think there are the following ways to solve this problem:
> 1、Increases the corePoolSize, but increases CPU consumption.
> 2、Use CachedThreadPool which uses SynchronousQueue as the workQueue, and the
> tasks execute immediately. But too many tasks can take up too many resources.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
