debug

Author: xxntti3n

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

@@ -305,15 +305,32 @@ private void startAsynchronouslySplit() {
     private void splitTable(TableId nextTable) {
         LOG.info("Start splitting table {} into chunks...", nextTable);
         long start = System.currentTimeMillis();
+        long lastProgressTime = System.currentTimeMillis();
         int chunkNum = 0;
         boolean hasRecordSchema = false;
         // split the given table into chunks (snapshot splits)
         do {
             synchronized (lock) {
                 List<MySqlSnapshotSplit> splits;
                 try {
+                    // Add progress logging
+                    long now = System.currentTimeMillis();
+                    if (now - lastProgressTime > 60000L) { // Log every minute
+                        LOG.info(
+                                "Still splitting table {}, chunks generated so far: {}, elapsed time: {}ms",
+                                nextTable,
+                                chunkNum,
+                                now - start);
+                        lastProgressTime = now;
+                    }
+
                     splits = chunkSplitter.splitChunks(partition, nextTable);
                 } catch (Exception e) {
+                    LOG.error(
+                            "Error when splitting chunks for table {}: {}",
+                            nextTable,
+                            e.getMessage(),
+                            e);
                     throw new IllegalStateException(
                             "Error when splitting chunks for " + nextTable, e);
                 }
@@ -367,8 +384,20 @@ public Optional<MySqlSplit> getNext() {
                         split.toMySqlSnapshotSplit(tableSchemas.get(split.getTableId())));
             } else if (!remainingTables.isEmpty()) {
                 try {
-                    // wait for the asynchronous split to complete
-                    lock.wait();
+                    // wait for the asynchronous split to complete with timeout
+                    final long timeout = 30000L; // 30 seconds timeout
+                    lock.wait(timeout);
+
+                    // Check if we're still waiting after timeout
+                    if (remainingSplits.isEmpty() && !remainingTables.isEmpty()) {
+                        LOG.warn(
+                                "Timeout waiting for asynchronous split generation, remaining tables: {}",
+                                remainingTables);
+                        // Try to restart the splitting process
+                        startAsynchronouslySplit();
+                        // Wait again with shorter timeout
+                        lock.wait(5000L);
+                    }
                 } catch (InterruptedException e) {
                     throw new FlinkRuntimeException(
                             "InterruptedException while waiting for asynchronously snapshot split");

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java

@@ -236,6 +236,7 @@ private void assignSplits() {
                 break;
             }
         }
+        LOG.info("The enumerator finishes this round of assignment.");
     }
 
     private boolean shouldCloseIdleReader(int nextAwaiting) {