[AlexYinHan]

Purpose of the change

This PR addresses FLINK-41061.

The adaptive partition benchmark deliberately slows selected downstream subtasks to measure whether adaptive partition selection can reduce the impact of processing skew. Currently, these subtasks sleep for 1 ms on every record, making this benchmark run for hours and keeps failing the daily run.

Brief change log

In SkewableHigherMultiply.java, sleep for 1 ms every 100 records, rather than every record, using an independent counter in each slow operator instance.

Verifying this change

Environment and method

• macOS ARM64, Azul Zulu JDK 11.0.26, JMH 1.37.
• Flink 855597a5460630eb8f72510277bfed7c0be0a004 (2.4-SNAPSHOT).
• Benchmark baseline: 123532a4a510002f52f2b5dff42541958e0943ac; the modified variant differs only by this PR's sleep-frequency change.

The following JMH command was used for the before/after comparison, with BENCHMARK_CLASSPATH containing the compiled harness and its dependencies:

"$JAVA_HOME/bin/java" -cp "$BENCHMARK_CLASSPATH" org.openjdk.jmh.Main \
  org.apache.flink.benchmark.AdaptivePartitionThroughputBenchmark.mapRebalanceMapRescaleMapSink \
  -f 1 -wi 0 -i 1 -r 1s -t 1 -to 10m \
  -p adaptivePartitioner=false \
  -p adaptivePartitionerMaxTraverseSize=2 \
  -p sourceType=F27_UNBOUNDED \
  -foe true -rf json -rff result.json

The same command was repeated for (adaptivePartitioner, adaptivePartitionerMaxTraverseSize) = (true, 2) and (true, 6), saving each result separately. (false, 6) was not run locally because the traverse-size setting is unused when adaptive partitioning is disabled.

Results

With 1 fork, no warmup and 1 measurement iteration, the average invocation times were:

Adaptive partitioner	Max traverse size	Before: sleep every record	After: sleep every 100 records
Disabled	2	220.0 s	2.38 s
Enabled	2	58.2 s	1.10 s
Enabled	6	39.6 s	0.74 s

These are average times per benchmark invocation, calculated from JMH throughput and @OperationsPerInvocation(150_000), not total JVM process durations or full CI runtimes.

As an additional check, the modified variant was run with 1 fork, 1 warmup iteration and 3 measurement iterations, replacing -wi 0 -i 1 with -wi 1 -w 1s -i 3. The per-iteration average invocation times were:

Adaptive partitioner	Max traverse size	Average invocation time in each of the three iterations
Disabled	2	2.027 s, 1.939 s, 1.943 s
Enabled	2	0.645 s, 0.564 s, 0.590 s
Enabled	6	0.409 s, 0.392 s, 0.378 s

All runs completed successfully. The modified benchmark still showed an adaptive-partitioner benefit in these local measurements, while substantially reducing execution time.

Impact and documentation

• The benchmark continues to introduce processing skew, but its intensity is reduced. Results before and after this change are not directly comparable as a Flink performance trend; this change establishes a new workload baseline.
• Only the adaptive partition benchmark currently uses the modified helper, so the change applies to all four parameter combinations without changing other benchmarks.
• No Flink production code, dependencies or public APIs are changed. No separate documentation changes are included.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: [Qoder Sonus]

[RocMarshal]

Could we split this attribute into two for consise semantic :

• additionalCostMillisPerBatch
• recordsOfAdditionalCostMillis

WDYTA?

[RocMarshal]

Thanks @AlexYinHan

Could we split this attribute into two for consise semantic :

• additionalCostMillisPerBatch
• recordsOfAdditionalCostMillis

WDYTA?

[AlexYinHan]

@RocMarshal Good point. I didn't pay much attention to the semantics.
I'm just not sure about the meaning of recordsOfAdditionalCostMillis. Could it be recordsPerBatch? I can then rename recordsSinceSleep to numRecordsForCurrentBatch for consistency.

[RocMarshal]

Hi, @AlexYinHan
Sorry for my Ambiguous semantics mentioned above:

Yes. Just adding a few minor changes.

• private final int additionalCostMillisPerRecord; -> private final int additionalCostMillisPerBatch;

The overview changed lines as:

private final int additionalCostMillisPerBatch;
private final int recordsPerBatch;
private int numRecordsForCurrentBatch;

Hope it helps a little.

[RocMarshal]

@AlexYinHan Nice work!

LGTM +1.

[MartijnVisser]

This might need a rebase from master to get CI to pass

[RocMarshal]

Hi, @AlexYinHan Could you help rebase the latest master branch then re-push it ?
Thx

[AlexYinHan]

Hi, @MartijnVisser @RocMarshal . Thanks for the information. I have rebased the latest master branch.

[RocMarshal]

Thanks @AlexYinHan for the patch and @MartijnVisser for the help.

The CI was successful.

Merging...