From 3bb532bfa194df1b2570a49b2ee6e12a40ea0e4b Mon Sep 17 00:00:00 2001 From: Zakelly Date: Fri, 27 Sep 2024 15:04:17 +0800 Subject: [PATCH] [hotfix][state/forst] Use correct executor for ForSt coordinator --- .../java/org/apache/flink/state/forst/ForStStateExecutor.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java index 3f17edebc4ce85..ebbf211e5ea5e2 100644 --- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java +++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java @@ -85,7 +85,7 @@ public ForStStateExecutor( this.coordinatorThread = coordinatorInline ? org.apache.flink.util.concurrent.Executors.newDirectExecutorService() - : Executors.newSingleThreadScheduledExecutor( + : Executors.newSingleThreadExecutor( new ExecutorThreadFactory( "ForSt-StateExecutor-Coordinator-And-Write")); this.readThreadCount = readIoParallelism; @@ -101,7 +101,7 @@ public ForStStateExecutor( this.coordinatorThread = coordinatorInline ? org.apache.flink.util.concurrent.Executors.newDirectExecutorService() - : Executors.newSingleThreadScheduledExecutor( + : Executors.newSingleThreadExecutor( new ExecutorThreadFactory("ForSt-StateExecutor-Coordinator")); if (readIoParallelism <= 0 || writeIoParallelism <= 0) { this.readThreadCount = Math.max(readIoParallelism, writeIoParallelism);