add log for enumerator

Author: LiebingYu

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitGenerator.java

@@ -32,6 +32,9 @@
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.utils.ExceptionUtils;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
@@ -52,6 +55,8 @@
 /** A generator for lake splits. */
 public class LakeSplitGenerator {
 
+    private static final Logger LOG = LoggerFactory.getLogger(LakeSplitGenerator.class);
+
     private final TableInfo tableInfo;
     private final Admin flussAdmin;
     private final OffsetsInitializer.BucketOffsetsRetriever bucketOffsetsRetriever;
@@ -88,6 +93,7 @@ public List<SourceSplitBase> generateHybridLakeFlussSplits() throws Exception {
         LakeSnapshot lakeSnapshotInfo;
         try {
             lakeSnapshotInfo = flussAdmin.getLatestLakeSnapshot(tableInfo.getTablePath()).get();
+            LOG.info("Get lake snapshot info: {}", lakeSnapshotInfo);
         } catch (Exception exception) {
             if (ExceptionUtils.stripExecutionException(exception)
                     instanceof LakeTableSnapshotNotExistException) {
@@ -105,6 +111,7 @@ public List<SourceSplitBase> generateHybridLakeFlussSplits() throws Exception {
                                 .createPlanner(
                                         (LakeSource.PlannerContext) lakeSnapshotInfo::getSnapshotId)
                                 .plan());
+        LOG.info("Group lake splits: {}", lakeSplits);
 
         Map<TableBucket, Long> tableBucketsOffset = lakeSnapshotInfo.getTableBucketsOffset();
         if (isPartitioned) {

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java

@@ -258,6 +258,7 @@ public void start() {
                     // we'll need to consider lake splits
                     List<SourceSplitBase> hybridLakeFlussSplits = generateHybridLakeFlussSplits();
                     if (hybridLakeFlussSplits != null) {
+                        LOG.info("Generated hybrid lake splits: {}", hybridLakeFlussSplits);
                         // handle hybrid lake fluss splits firstly
                         handleSplitsAdd(hybridLakeFlussSplits, null);
                     }
@@ -616,6 +617,7 @@ private List<SourceSplitBase> generateHybridLakeFlussSplits() {
         // should be restored from checkpoint, shouldn't
         // list splits again
         if (pendingHybridLakeFlussSplits != null) {
+            LOG.info("Still have pending lake fluss splits, shouldn't list splits again.");
             return pendingHybridLakeFlussSplits;
         }
         try {
@@ -659,6 +661,7 @@ private void handlePartitionsRemoved(Collection<Partition> removedPartitionInfo)
                                 split ->
                                         removedPartitionsMap.containsKey(
                                                 split.getTableBucket().getPartitionId())));
+        LOG.info("Removed partitions {}", removedPartitionsMap);
 
         // send partition removed event to all readers
         PartitionsRemovedEvent event = new PartitionsRemovedEvent(removedPartitionsMap);
@@ -709,6 +712,8 @@ private void addSplitToPendingAssignments(Collection<SourceSplitBase> newSplits)
     private void assignPendingSplits(Set<Integer> pendingReaders) {
         Map<Integer, List<SourceSplitBase>> incrementalAssignment = new HashMap<>();
 
+        LOG.info("Assigning pending splits to readers {}", pendingReaders);
+        LOG.info("Pending split assignment {}", pendingSplitAssignment);
         // Check if there's any pending splits for given readers
         for (int pendingReader : pendingReaders) {
             checkReaderRegistered(pendingReader);
@@ -758,6 +763,8 @@ private void assignPendingSplits(Set<Integer> pendingReaders) {
         if (!incrementalAssignment.isEmpty()) {
             LOG.info("Assigning splits to readers {}", incrementalAssignment);
             context.assignSplits(new SplitsAssignment<>(incrementalAssignment));
+        } else {
+            LOG.info("No pending splits to assign to readers {}", pendingReaders);
         }
 
         if (noMoreNewSplits) {

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplit.java

@@ -72,4 +72,14 @@ public DataSplit dataSplit() {
     public boolean isBucketUnAware() {
         return isBucketUnAware;
     }
+
+    @Override
+    public String toString() {
+        return "PaimonSplit{"
+                + "dataSplit="
+                + dataSplit
+                + ", isBucketUnAware="
+                + isBucketUnAware
+                + '}';
+    }
 }