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 long estimateBoundedSourceSize( @@ -45,13 +49,25 @@ static ArrayList> splitBoundedSource( throws Exception { long desiredSizeBytes = getDesiredSizeBytes(boundedSource, pipelineOptions, numSplits, estimatedSizeBytes); - return toFlinkSplits(boundedSource.split(desiredSizeBytes, pipelineOptions)); + List> 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 ArrayList> splitUnboundedSource( UnboundedSource unboundedSource, PipelineOptions pipelineOptions, int numSplits) throws Exception { - return toFlinkSplits(unboundedSource.split(numSplits, pipelineOptions)); + List> 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 selectAndSplit() throws Exception { ArrayList> 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); }