AlexYinHan opened a new pull request, #120:
URL: https://github.com/apache/flink-benchmarks/pull/120
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing to Apache Flink benchmarks.
Please follow the Flink contribution guide:
https://flink.apache.org/how-to-contribute/contribute-code/
-->
## Purpose of the change
This PR addresses
[FLINK-41061](https://issues.apache.org/jira/browse/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:
```bash
"$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?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change the checkbox below to `[X]` and replace the placeholder in the
"Generated-by"
line with the tool name and version. Otherwise remove the "Generated-by"
line.
See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this
PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed.
See
https://flink.apache.org/how-to-contribute/contribute-code/#ai-assisted-contributions
for the full guidance.
-->
- [x] Yes (please specify the tool below)
Generated-by: [Qoder Sonus]
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
