Repository: apache/flink
File: flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleWithMixedExchangesITCase.java (Java)

Code before:
                .keyBy((KeySelector<Long, Long>) value -> value)
                .map(
                        x -> {
                            Thread.sleep(5);
                            return x;
                        })
                .name("MapAfterKeyBy")
                .setParallelism(getRandomParallelism());

        return env.executeAsync();
    }

    private static int getRandomParallelism() {
        return RANDOM.nextInt(MAX_SLOTS) + 1;
    }

    private static DataGeneratorSource<Long> createSource() {
        return new DataGeneratorSource<>(
                index -> index, Long.MAX_VALUE, RateLimiterStrategy.perSecond(5000), Types.LONG) {

            @Override
            public SplitEnumerator<NumberSequenceSplit, Collection<NumberSequenceSplit>>
                    createEnumerator(SplitEnumeratorContext<NumberSequenceSplit> enumContext) {
                // force large number of splits to avoid sub-tasks finishing prematurely
                // and failing the test assumption of all sub-tasks always running
                return super.createEnumerator(new MockSplitEnumeratorContext<>(MAX_SLOTS));
            }
        };
    }

    /** A simple CoMapFunction that sleeps for 1ms for each element. */
    private static class SleepingCoMap implements CoMapFunction<Long, Long, Long> {
        @Override
        public Long map1(Long value) throws Exception {
            Thread.sleep(1);
            return value;
        }

        @Override
        public Long map2(Long value) throws Exception {
            Thread.sleep(1);
            return value;
        }
    }

    public interface ExecuteJobViaEnv {
        JobClient executeJob(StreamExecutionEnvironment env) throws Exception;
    }
}


Review comment (suggestion):
is it possible to only change the split number but parallelism?

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

Code after:
                            Thread.sleep(5);
                            return x;
                        })
                .name("MapAfterKeyBy")
                .setParallelism(getRandomParallelism());

        return env.executeAsync();
    }

    private static int getRandomParallelism() {
        return RANDOM.nextInt(MAX_SLOTS) + 1;
    }

    private static DataGeneratorSource<Long> createSource() {
        return new DataGeneratorSource<>(
                index -> index,
                new NumberSequenceSource(0, Long.MAX_VALUE - 1) {
                    @Override
                    protected List<NumberSequenceSplit> splitNumberRange(
                            long from, long to, int numSplitsIgnored) {
                        return super.splitNumberRange(from, to, MAX_SLOTS);
                    }
                },
                RateLimiterStrategy.perSecond(5000),
                Types.LONG) {};
    }

    /** A simple CoMapFunction that sleeps for 1ms for each element. */
    private static class SleepingCoMap implements CoMapFunction<Long, Long, Long> {
        @Override
        public Long map1(Long value) throws Exception {
            Thread.sleep(1);
            return value;
        }

        @Override
        public Long map2(Long value) throws Exception {
            Thread.sleep(1);
            return value;
        }
    }

    public interface ExecuteJobViaEnv {
        JobClient executeJob(StreamExecutionEnvironment env) throws Exception;
    }
}