This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 267ea2b3c836e4a59e7dc382707cf212787181ef Author: Hanyu (Peter) Zheng <[email protected]> AuthorDate: Wed Oct 28 09:12:26 2026 -0500 [FLINK-41217][core] Support empty sequences in NumberSequenceSource A NumberSequenceSource from 'from' to 'from' - 1 is empty, so that a DataGeneratorSource can generate no data points. --- .../connector/source/lib/NumberSequenceSource.java | 9 +++++++-- .../source/lib/NumberSequenceSourceTest.java | 23 ++++++++++++++++++++++ 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/source/lib/NumberSequenceSource.java b/flink-core/src/main/java/org/apache/flink/api/connector/source/lib/NumberSequenceSource.java index 53e6a2832b7..c455c4b5a02 100644 --- a/flink-core/src/main/java/org/apache/flink/api/connector/source/lib/NumberSequenceSource.java +++ b/flink-core/src/main/java/org/apache/flink/api/connector/source/lib/NumberSequenceSource.java @@ -41,6 +41,7 @@ import org.apache.flink.util.NumberSequenceIterator; import java.io.IOException; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.List; import static org.apache.flink.util.Preconditions.checkArgument; @@ -76,10 +77,11 @@ public class NumberSequenceSource /** * Creates a new {@code NumberSequenceSource} that produces parallel sequences covering the - * range {@code from} to {@code to} (both boundaries are inclusive). + * range {@code from} to {@code to} (both boundaries are inclusive). The sequence is empty if + * {@code to} is {@code from - 1}. */ public NumberSequenceSource(long from, long to) { - checkArgument(from <= to, "'from' must be <= 'to'"); + checkArgument(from <= to || to == from - 1, "'to' must be >= 'from' - 1"); this.from = from; this.to = to; } @@ -139,6 +141,9 @@ public class NumberSequenceSource } protected List<NumberSequenceSplit> splitNumberRange(long from, long to, int numSplits) { + if (from > to) { + return Collections.emptyList(); + } final NumberSequenceIterator[] subSequences = new NumberSequenceIterator(from, to).split(numSplits); final ArrayList<NumberSequenceSplit> splits = new ArrayList<>(subSequences.length); diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceTest.java b/flink-core/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceTest.java index 4aee5a82761..d708bef61b2 100644 --- a/flink-core/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceTest.java +++ b/flink-core/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceTest.java @@ -24,6 +24,8 @@ import org.apache.flink.api.connector.source.SourceEvent; import org.apache.flink.api.connector.source.SourceOutput; import org.apache.flink.api.connector.source.SourceReader; import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.io.InputStatus; import org.apache.flink.metrics.groups.SourceReaderMetricGroup; @@ -35,8 +37,11 @@ import org.junit.jupiter.api.Test; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.List; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.fail; /** Tests for the {@link NumberSequenceSource}. */ @@ -78,6 +83,24 @@ class NumberSequenceSourceTest { validateSequence(result, from, to); } + @Test + void testEmptySequence() throws Exception { + final NumberSequenceSource source = new NumberSequenceSource(5, 4); + + final SplitEnumerator< + NumberSequenceSource.NumberSequenceSplit, + Collection<NumberSequenceSource.NumberSequenceSplit>> + enumerator = source.createEnumerator(new MockSplitEnumeratorContext<>(2)); + + assertThat(enumerator.snapshotState(1L)).isEmpty(); + } + + @Test + void testToBeforeFromMinusOne() { + assertThatThrownBy(() -> new NumberSequenceSource(5, 3)) + .isInstanceOf(IllegalArgumentException.class); + } + private static void validateSequence( final List<Long> sequence, final long from, final long to) { if (sequence.size() != to - from + 1) {
