[
https://issues.apache.org/jira/browse/FLINK-37317?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jufang He updated FLINK-37317:
------------------------------
Description:
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}
result:
{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.
was:
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}
result:
{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.
> 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}
> result:
> {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)
