fhan688 opened a new pull request, #20189:
URL: https://github.com/apache/hudi/pull/20189
### Describe the issue this Pull Request addresses
Flink distributes the job-wide `write.rate.limit` across write subtasks
using integer division:
```text
rateLimitPerSubtask = write.rate.limit / parallelism
```
When `write.rate.limit` is lower than the write parallelism, the
per-subtask value becomes zero. This creates a `RateLimiter` backed by a
zero-permit semaphore. Its scheduled refill also releases zero permits, so the
first `acquire(1)` blocks indefinitely and the Flink pipeline stops making
progress.
In addition, both rate-limited Flink write functions create a scheduled
executor during `open()`, but do not stop it during `close()`. Task shutdown,
restart, or recovery can therefore leave rate-limiter scheduler threads running.
### Summary and Changelog
Allow positive write rate limits lower than the Flink write parallelism
without blocking the pipeline, and reliably release rate-limiter resources.
- Add support for configurable permit refill periods to the common
`RateLimiter`.
- Reject zero permits, invalid refill periods, and invalid acquire
requests instead of allowing permanently blocked limiters.
- Make `RateLimiter.stop()` thread-safe and idempotent.
- Allow blocked permit acquisition to terminate after the limiter is
stopped.
- Preserve the thread interruption status when permit acquisition is
interrupted.
- Use a named daemon thread for the rate-limiter scheduler as an
additional lifecycle safeguard.
- Add a shared Flink rate-limiter factory to distribute a job-wide rate
across subtasks.
- For `write.rate.limit >= parallelism`, preserve the existing per-second
permit allocation.
- For `write.rate.limit < parallelism`, allocate one permit per calculated
multi-second or sub-second interval instead of creating a zero-permit limiter.
- Stop the limiter from `RowDataToHoodieFunctionWithRateLimit.close()` and
`AppendWriteFunctionWithRateLimit.close()`, including when parent resource
cleanup throws.
- Add regression tests covering low rate limits, custom refill periods,
invalid arguments, blocked acquisition shutdown, repeated closure, and both
Flink write paths.
No code was copied from external sources.
### Impact
- Positive `write.rate.limit` values lower than the Flink write
parallelism no longer cause the write pipeline to stall permanently.
- The configured job-wide limit remains conservatively distributed across
subtasks; this fix does not use a minimum one-per-second allocation that could
raise the aggregate rate above the configured value.
- Rate-limiter scheduler threads are terminated when their owning Flink
functions close.
- Existing behavior is preserved when the configured rate is greater than
or equal to the write parallelism.
- An additive `RateLimiter` factory overload is introduced for
configurable refill periods.
- No configuration defaults, storage formats, dependencies, or user-facing
method signatures are changed.
### Risk Level
medium
This change modifies the common rate-limiter lifecycle and Flink write
throttling behavior.
The following validation was performed using JDK 11 and the Flink 1.20
build profile:
```bash
mvn -pl hudi-common -am \
-DskipITs \
-Dtest=TestRatelimiter \
-DfailIfNoTests=false \
-Dsurefire.failIfNoSpecifiedTests=false \
test-compile surefire:test
```
```bash
mvn -Pflink1.20 \
-pl hudi-flink-datasource/hudi-flink -am \
-DskipITs \
-Dtest=TestFlinkRateLimiterFactory,TestRowDataToHoodieFunctionWithRateLimit,TestAppendWriteFunctionWithRateLimit,TestTransformUtilities,TestAppendWriteFunctionWithBufferSort
\
-DfailIfNoTests=false \
-Dsurefire.failIfNoSpecifiedTests=false \
test-compile surefire:test
```
Results:
- `TestRatelimiter`: 5 tests passed.
- Flink rate-limiter and related regression tests: 28 tests passed.
- No failures, errors, or skipped tests.
- Compilation, Checkstyle, and license checks passed.
- `git diff --check` passed.
- The full project test suite was not run.
### Documentation Update
none
This fixes existing rate-limiter behavior without adding or changing
user-facing configuration.
### Contributor's checklist
- [ ] Read through [contributor's
guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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]
