Fix meaningless thread creation every time checkpoint mysql load

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -1574,6 +1574,8 @@ private void startMasterOnlyDaemonThreads() {
 
     // start threads that should running on all FE
     private void startNonMasterDaemonThreads() {
+        // start load manager thread
+        loadManager.start();
         tabletStatMgr.start();
         // load and export job label cleaner thread
         labelCleaner.start();

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java

@@ -104,6 +104,11 @@ public LoadManager(LoadJobScheduler loadJobScheduler) {
         this.mysqlLoadManager = new MysqlLoadManager(tokenManager);
     }
 
+    public void start() {
+        tokenManager.start();
+        mysqlLoadManager.start();
+    }
+
     /**
      * This method will be invoked by the broker load(v2) now.
      */

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/MysqlLoadManager.java

@@ -73,7 +73,7 @@
 public class MysqlLoadManager {
     private static final Logger LOG = LogManager.getLogger(MysqlLoadManager.class);
 
-    private final ThreadPoolExecutor mysqlLoadPool;
+    private  ThreadPoolExecutor mysqlLoadPool;
     private final TokenManager tokenManager;
 
     private static class MySqlLoadContext {
@@ -138,16 +138,19 @@ public boolean isExpired() {
     }
 
     private final Map<String, MySqlLoadContext> loadContextMap = new ConcurrentHashMap<>();
-    private final EvictingQueue<MySqlLoadFailRecord> failedRecords;
+    private  EvictingQueue<MySqlLoadFailRecord> failedRecords;
     private ScheduledExecutorService periodScheduler = Executors.newScheduledThreadPool(1,
             new CustomThreadFactory("mysql-load-fail-record-cleaner"));
 
     public MysqlLoadManager(TokenManager tokenManager) {
+        this.tokenManager = tokenManager;
+    }
+
+    public void start() {
         int poolSize = Config.mysql_load_thread_pool;
         // MySqlLoad pool can accept 4 + 4 * 5 = 24  requests by default.
         this.mysqlLoadPool = ThreadPoolManager.newDaemonFixedThreadPool(poolSize, poolSize * 5,
                 "Mysql Load", true);
-        this.tokenManager = tokenManager;
         this.failedRecords = EvictingQueue.create(Config.mysql_load_in_memory_record);
         this.periodScheduler.scheduleAtFixedRate(this::cleanFailedRecords, 1, 24, TimeUnit.HOURS);
     }

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/TokenManager.java

@@ -42,11 +42,14 @@
 public class TokenManager {
     private static final Logger LOG = LogManager.getLogger(TokenManager.class);
 
-    private final int thriftTimeoutMs = 300 * 1000;
-    private final EvictingQueue<String> tokenQueue;
-    private final ScheduledExecutorService tokenGenerator;
+    private  int thriftTimeoutMs = 300 * 1000;
+    private  EvictingQueue<String> tokenQueue;
+    private  ScheduledExecutorService tokenGenerator;
 
     public TokenManager() {
+    }
+
+    public void start() {
         this.tokenQueue = EvictingQueue.create(Config.token_queue_size);
         // init one token to avoid async issue.
         this.tokenQueue.offer(generateNewToken());