[flink]fix and add log

Author: Alibaba-HZY

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

@@ -368,7 +368,7 @@ private List<PartitionInfo> applyPartitionFilter(List<PartitionInfo> partitionIn
                                                     convertPartitionInfoToInternalRow(
                                                             partitionInfo)))
                             .collect(Collectors.toList());
-            
+
             int filteredSize = filteredPartitionInfos.size();
             // Only log when there's actual filtering happening or when it's the first time
             if (originalSize != filteredSize) {
@@ -411,12 +411,14 @@ private void checkPartitionChanges(Set<PartitionInfo> partitionInfos, Throwable
             LOG.error("Failed to list partitions for {}", tablePath, t);
             return;
         }
-        
+
         if (LOG.isDebugEnabled()) {
-            LOG.debug("Checking partition changes for table {}, found {} partitions", 
-                    tablePath, partitionInfos.size());
+            LOG.debug(
+                    "Checking partition changes for table {}, found {} partitions",
+                    tablePath,
+                    partitionInfos.size());
         }
-        
+
         final PartitionChange partitionChange = getPartitionChange(partitionInfos);
         if (partitionChange.isEmpty()) {
             if (LOG.isDebugEnabled()) {
@@ -427,17 +429,24 @@ private void checkPartitionChanges(Set<PartitionInfo> partitionInfos, Throwable
 
         // handle removed partitions
         if (!partitionChange.removedPartitions.isEmpty()) {
-            LOG.info("Handling {} removed partitions for table {}: {}", 
-                    partitionChange.removedPartitions.size(), tablePath, partitionChange.removedPartitions);
+            LOG.info(
+                    "Handling {} removed partitions for table {}: {}",
+                    partitionChange.removedPartitions.size(),
+                    tablePath,
+                    partitionChange.removedPartitions);
             handlePartitionsRemoved(partitionChange.removedPartitions);
         }
 
         // handle new partitions
         if (!partitionChange.newPartitions.isEmpty()) {
-            LOG.info("Handling {} new partitions for table {}: {}", 
-                    partitionChange.newPartitions.size(), tablePath, partitionChange.newPartitions);
+            LOG.info(
+                    "Handling {} new partitions for table {}: {}",
+                    partitionChange.newPartitions.size(),
+                    tablePath,
+                    partitionChange.newPartitions);
             context.callAsync(
-                    () -> initPartitionedSplits(partitionChange.newPartitions), this::handleSplitsAdd);
+                    () -> initPartitionedSplits(partitionChange.newPartitions),
+                    this::handleSplitsAdd);
         }
     }