[FLINK-38613][tests] Force number of splits in UnalignedCheckpointRescaleWithMixedExchangesITCase

[rkhachatryan]

Motivation: avoid sub-tasks finishing prematurely and fail the test assumption of all sub-tasks always running

[flinkbot]

CI report:

• 8e01e8e Azure: FAILURE
• b0e9e90 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[1996fanrui]

Thanks for the analysis. I reproduced and investigated it further.

The root cause is: on restore, the available splits come from the checkpoint state, not from the new parallelism. For example, first run with parallelism 3 → NumberSequenceSource creates 3 splits, all assigned immediately, checkpoint saves 0 remaining splits. Restore with parallelism 7 → restoreEnumerator() gets 0 splits from checkpoint, so 4 extra subtasks get NoMoreSplits and finish.

I like your Option 3 that mentioned in https://issues.apache.org/jira/browse/FLINK-38613, but scoped to the test only — no production code changes. Subclassed DataGeneratorSource and overrode createEnumerator() to always create at least MAX_SLOTS splits:

private static class TestDataGeneratorSource extends DataGeneratorSource<Long> {
    TestDataGeneratorSource() {
        super(index -> index, Long.MAX_VALUE, RateLimiterStrategy.perSecond(5000), Types.LONG);
    }

    @Override
    public SplitEnumerator<NumberSequenceSplit, Collection<NumberSequenceSplit>>
            createEnumerator(SplitEnumeratorContext<NumberSequenceSplit> enumContext) {
        NumberSequenceSource source =
                new NumberSequenceSource(0, Long.MAX_VALUE - 1) {
                    @Override
                    protected List<NumberSequenceSplit> splitNumberRange(
                            long from, long to, int numSplits) {
                        return super.splitNumberRange(
                                from, to, Math.max(numSplits, MAX_SLOTS));
                    }
                };
        return source.createEnumerator(enumContext);
    }
}

WDYT?

[1996fanrui]

is it possible to only change the split number but parallelism?

This ITCase is testing the rescale case, fixed parallelism covers fewer cases.

[rkhachatryan]

I don't think that SplitEnumerator is responsible for parallelism, is it?

[1996fanrui]

I think you are right. My concern is it will affect the semantics, the input parameter of MockSplitEnumeratorContext is parallelism, and interface name of SplitEnumeratorContext is currentParallelism. It might caused potential bug in the future.

For example, the parallelism will be used in other places as well, but our intention is that only changing split number.