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,13 +384,63 @@ 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 timed out and still have no splits
+                    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);
+
+                        // After restart and wait, check again for splits
+                        if (!remainingSplits.isEmpty()) {
+                            // Splits are now available after restart
+                            Iterator<MySqlSchemalessSnapshotSplit> iterator =
+                                    remainingSplits.iterator();
+                            MySqlSchemalessSnapshotSplit split = iterator.next();
+                            remainingSplits.remove(split);
+                            assignedSplits.put(split.splitId(), split);
+                            addAlreadyProcessedTablesIfNotExists(split.getTableId());
+                            return Optional.of(
+                                    split.toMySqlSnapshotSplit(
+                                            tableSchemas.get(split.getTableId())));
+                        } else if (!remainingTables.isEmpty()) {
+                            // Still no splits after restart, but we have tables - recurse normally
+                            return getNext();
+                        } else {
+                            // No tables left after restart
+                            closeExecutorService();
+                            return Optional.empty();
+                        }
+                    }
+                    // If we didn't timeout (normal case), check if splits are available now
+                    if (!remainingSplits.isEmpty()) {
+                        Iterator<MySqlSchemalessSnapshotSplit> iterator =
+                                remainingSplits.iterator();
+                        MySqlSchemalessSnapshotSplit split = iterator.next();
+                        remainingSplits.remove(split);
+                        assignedSplits.put(split.splitId(), split);
+                        addAlreadyProcessedTablesIfNotExists(split.getTableId());
+                        return Optional.of(
+                                split.toMySqlSnapshotSplit(tableSchemas.get(split.getTableId())));
+                    } else if (!remainingTables.isEmpty()) {
+                        // No splits yet but still have tables to process - recurse
+                        return getNext();
+                    } else {
+                        // No tables left
+                        closeExecutorService();
+                        return Optional.empty();
+                    }
                 } catch (InterruptedException e) {
                     throw new FlinkRuntimeException(
                             "InterruptedException while waiting for asynchronously snapshot split");
                 }
-                return getNext();
             } else {
                 closeExecutorService();
                 return Optional.empty();

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

@@ -202,6 +202,7 @@ public void close() throws IOException {
 
     private void assignSplits() {
         final Iterator<Integer> awaitingReader = readersAwaitingSplit.iterator();
+        boolean hasAssignmentFailure = false;
 
         while (awaitingReader.hasNext()) {
             int nextAwaiting = awaitingReader.next();
@@ -230,12 +231,26 @@ private void assignSplits() {
                 }
                 awaitingReader.remove();
                 LOG.info("The enumerator assigns split {} to subtask {}", mySqlSplit, nextAwaiting);
+                hasAssignmentFailure = false; // Reset on successful assignment
             } else {
-                // there is no available splits by now, skip assigning
+                LOG.warn(
+                        "The enumerator has no more splits to assign to subtask {} by now. This may indicate a split generation timeout or all splits are exhausted.",
+                        nextAwaiting);
+                // Don't break immediately - try other readers first
+                hasAssignmentFailure = true;
                 requestBinlogSplitUpdateIfNeed();
-                break;
             }
         }
+
+        // If we had assignment failures and there are still readers waiting,
+        // schedule another assignment attempt to handle potential async split generation
+        if (hasAssignmentFailure && !readersAwaitingSplit.isEmpty()) {
+            LOG.info(
+                    "Some readers couldn't get splits due to temporary unavailability. Scheduling retry...");
+            // The async check interval will trigger another assignSplits() call
+        }
+
+        LOG.info("The enumerator finishes this round of assignment.");
     }
 
     private boolean shouldCloseIdleReader(int nextAwaiting) {