fix

Author: JNSimba

fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoad.java

@@ -38,6 +38,7 @@
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -80,6 +81,7 @@ public class DorisBatchStreamLoad implements Serializable {
     private BlockingQueue<BatchRecordBuffer> flushQueue;
     private final AtomicBoolean started;
     private volatile boolean loadThreadAlive = false;
+    private final CountDownLatch loadThreadStarted = new CountDownLatch(1);
     private AtomicReference<Throwable> exception = new AtomicReference<>(null);
     private long maxBlockedBytes;
     private final AtomicLong currentCacheBytes = new AtomicLong(0L);
@@ -110,6 +112,16 @@ public DorisBatchStreamLoad(long jobId, String targetDb) {
         this.loadExecutorService.execute(loadAsyncExecutor);
         this.targetDb = targetDb;
         this.jobId = jobId;
+        // Wait for the load thread to start
+        try {
+            if (!loadThreadStarted.await(10, TimeUnit.SECONDS)) {
+                throw new RuntimeException("LoadAsyncExecutor thread startup timed out");
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(
+                    "Thread interrupted while waiting for load thread to start", e);
+        }
     }
 
     /**
@@ -310,6 +322,7 @@ public LoadAsyncExecutor(int flushQueueSize, long jobId) {
         public void run() {
             LOG.info("LoadAsyncExecutor start for jobId {}", jobId);
             loadThreadAlive = true;
+            loadThreadStarted.countDown();
             List<BatchRecordBuffer> recordList = new ArrayList<>(flushQueueSize);
             while (started.get()) {
                 recordList.clear();

fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java

@@ -166,8 +166,9 @@ public SplitReadResult readSplitRecords(JobBaseRecordRequest baseReq) throws Exc
             Fetcher<SourceRecords, SourceSplitBase> currentReader = this.getCurrentReader();
             if (currentReader == null || baseReq.isReload()) {
                 LOG.info(
-                        "No current reader or reload {}, create new split reader",
-                        baseReq.isReload());
+                        "No current reader or reload {}, create new split reader for job {}",
+                        baseReq.isReload(),
+                        baseReq.getJobId());
                 // build split
                 Tuple2<SourceSplitBase, Boolean> splitFlag = createSourceSplit(offsetMeta, baseReq);
                 split = splitFlag.f0;