[Flink] Fix flink source enum (lakesoul-io#573)

* add more log

Signed-off-by: chenxu <chenxu@dmetasoul.com>

* fix task assign logics

Signed-off-by: chenxu <chenxu@dmetasoul.com>

---------

Signed-off-by: chenxu <chenxu@dmetasoul.com>
Co-authored-by: chenxu <chenxu@dmetasoul.com>

Author: xuchen-plus

lakesoul-flink/src/main/java/org/apache/flink/lakesoul/source/LakeSoulAllPartitionDynamicSplitEnumerator.java

@@ -48,6 +48,7 @@ public class LakeSoulAllPartitionDynamicSplitEnumerator implements SplitEnumerat
     private long startTime;
     private long nextStartTime;
     private int hashBucketNum = -1;
+    String fullTableName;
 
     public LakeSoulAllPartitionDynamicSplitEnumerator(SplitEnumeratorContext<LakeSoulPartitionSplit> context, LakeSoulDynSplitAssigner splitAssigner, RowType rowType, long discoveryInterval, long startTime, String tableId, String hashBucketNum, List<String> partitionColumns, Plan partitionFilters) {
         this.context = context;
@@ -66,6 +67,9 @@ public LakeSoulAllPartitionDynamicSplitEnumerator(SplitEnumeratorContext<LakeSou
         this.partitionArrowSchema = new Schema(partitionFields);
         this.partitionFilters = partitionFilters;
         tableInfo = DataOperation.dbManager().getTableInfoByTableId(tableId);
+        fullTableName = tableInfo.getTableNamespace() + "." + tableInfo.getTableName();
+        LOG.info("Create Dyn enumerator for table name {}, tableId {}, context {}",
+                fullTableName, tableId, System.identityHashCode(context));
     }
 
     @Override
@@ -83,6 +87,8 @@ public synchronized void handleSplitRequest(int subtaskId, @Nullable String requ
         }
         int tasksSize = context.registeredReaders().size();
         if (tasksSize == 0) {
+            LOG.info("handleSplitRequest: Task size is 0 for subtaskId {} for table {}", subtaskId, fullTableName);
+            taskIdsAwaitingSplit.add(subtaskId);
             return;
         }
         Optional<LakeSoulPartitionSplit> nextSplit = this.splitAssigner.getNext(subtaskId, tasksSize);
@@ -134,10 +140,10 @@ private synchronized void processDiscoveredSplits(
         LOG.info("Process discovered splits {}, taskSize {}, oid {}, tid {}", splits,
                 tasksSize, System.identityHashCode(this),
                 Thread.currentThread().getId());
+        this.splitAssigner.addSplits(splits);
         if (tasksSize == 0) {
             return;
         }
-        this.splitAssigner.addSplits(splits);
         Iterator<Integer> iter = taskIdsAwaitingSplit.iterator();
         while (iter.hasNext()) {
             int taskId = iter.next();