fix UT

vernedeng · Aug 24, 2023 · bfb63e6 · bfb63e6
1 parent 3741e9b
commit bfb63e6
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/TopicManager.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/TopicManager.java
@@ -30,12 +30,10 @@ public abstract class TopicManager implements Cleanable {
 
     protected ClientContext context;
     protected QueryConsumeConfig queryConsumeConfig;
-    protected final ForkJoinPool pool;
 
     public TopicManager(ClientContext context, QueryConsumeConfig queryConsumeConfig) {
         this.context = context;
         this.queryConsumeConfig = queryConsumeConfig;
-        pool = new ForkJoinPool(context.config.getThreadPoolSize());
     }
 
     /**

diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongTopicManager.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongTopicManager.java
@@ -46,6 +46,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -65,6 +66,8 @@ public class InlongTopicManager extends TopicManager {
     private final Map<String, PulsarClient> pulsarClients = new ConcurrentHashMap<>();
     private final Map<String, TubeConsumerCreator> tubeFactories = new ConcurrentHashMap<>();
 
+    protected final ForkJoinPool pool;
+
     private volatile boolean stopAssign = false;
 
     private Collection<InLongTopic> assignedTopics;
@@ -73,6 +76,7 @@ public InlongTopicManager(ClientContext context, QueryConsumeConfig queryConsume
         super(context, queryConsumeConfig);
         executor.scheduleWithFixedDelay(this::updateMetaData, 0L,
                 context.getConfig().getUpdateMetaDataIntervalSec(), TimeUnit.SECONDS);
+        pool = new ForkJoinPool(context.getConfig().getThreadPoolSize());
     }
 
     @Override