From af5c38a8a0925cc5467c11f5d0853d569ab8a98e Mon Sep 17 00:00:00 2001
From: Paulius Kuzmickas <p.kuzmickas@gmail.com>
Date: Wed, 30 Sep 2026 11:45:08 +0100
Subject: [PATCH] [Flink] Restore source split logs

The move of source splitting into FlinkSourceSplitUtils dropped the
split-count logs from the default and lazy enumerators. Log in the
shared helpers so every enumerator reports how many splits a source
produced, and remove the now-duplicate size-based enumerator log.
---
 .../io/source/FlinkSourceSplitUtils.java      | 20 +++++++++++++++++--
 .../SizeBasedFlinkSourceSplitEnumerator.java  |  5 -----
 2 files changed, 18 insertions(+), 7 deletions(-)

diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceSplitUtils.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceSplitUtils.java
index dda99bf53bef..26a5dd4d9349 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceSplitUtils.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceSplitUtils.java
@@ -25,11 +25,15 @@
 import org.apache.beam.sdk.io.Source;
 import org.apache.beam.sdk.io.UnboundedSource;
 import org.apache.beam.sdk.options.PipelineOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** Shared Beam source sizing and splitting helpers. */
 final class FlinkSourceSplitUtils {
   static final long MEBIBYTE = 1024L * 1024L;
 
+  private static final Logger LOG = LoggerFactory.getLogger(FlinkSourceSplitUtils.class);
+
   private FlinkSourceSplitUtils() {}
 
   static <T> long estimateBoundedSourceSize(
@@ -45,13 +49,25 @@ static <T> ArrayList<FlinkSourceSplit<T>> splitBoundedSource(
       throws Exception {
     long desiredSizeBytes =
         getDesiredSizeBytes(boundedSource, pipelineOptions, numSplits, estimatedSizeBytes);
-    return toFlinkSplits(boundedSource.split(desiredSizeBytes, pipelineOptions));
+    List<? extends BoundedSource<T>> splits =
+        boundedSource.split(desiredSizeBytes, pipelineOptions);
+    LOG.info(
+        "Split bounded source {} in {} splits (estimated size {} bytes, "
+            + "desired split size {} bytes)",
+        boundedSource,
+        splits.size(),
+        estimatedSizeBytes,
+        desiredSizeBytes);
+    return toFlinkSplits(splits);
   }
 
   static <T> ArrayList<FlinkSourceSplit<T>> splitUnboundedSource(
       UnboundedSource<T, ?> unboundedSource, PipelineOptions pipelineOptions, int numSplits)
       throws Exception {
-    return toFlinkSplits(unboundedSource.split(numSplits, pipelineOptions));
+    List<? extends UnboundedSource<T, ?>> splits =
+        unboundedSource.split(numSplits, pipelineOptions);
+    LOG.info("Split source {} to {} splits", unboundedSource, splits);
+    return toFlinkSplits(splits);
   }
 
   static long getDesiredSizeBytes(
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/SizeBasedFlinkSourceSplitEnumerator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/SizeBasedFlinkSourceSplitEnumerator.java
index 78b0fb59f2c4..ff1a79e3c30a 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/SizeBasedFlinkSourceSplitEnumerator.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/SizeBasedFlinkSourceSplitEnumerator.java
@@ -131,11 +131,6 @@ private FlinkSourceEnumeratorState<T> selectAndSplit() throws Exception {
     ArrayList<FlinkSourceSplit<T>> splits =
         FlinkSourceSplitUtils.splitBoundedSource(
             boundedSource, pipelineOptions, numSplits, estimatedSizeBytes);
-    LOG.info(
-        "Split bounded source {} into {} splits using {} assignment",
-        boundedSource,
-        splits.size(),
-        selectedMode);
     return new FlinkSourceEnumeratorState<>(selectedMode, splits);
   }