You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "lvyanquan (via GitHub)" <gi...@apache.org> on 2025/10/21 12:52:17 UTC
[PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
lvyanquan opened a new pull request, #27134:
URL: https://github.com/apache/flink/pull/27134
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Provide a configuration solution for fine-grained read speed control in the Flink Source connector.
## Brief change log
- *Add new constructor for SourceReaderBase and SingleThreadMultiplexSourceReaderBase*
- *Add CompletionStage<Void> acquire(int requestSize) and void notifyStatusChange(S status) method in RateLimiter*
## Verifying this change
This change added tests and can be verified as follows:
*(example:)*
- *Added test that validates that Records were emitted with rate control*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (yes)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (JavaDocs)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2473357903
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java:
##########
@@ -27,14 +27,25 @@
/** The interface to rate limit execution of methods. */
@NotThreadSafe
@Experimental
-public interface RateLimiter {
+public interface RateLimiter<S> {
/**
* Returns a future that is completed once another event would not exceed the rate limit. For
* correct functioning, the next invocation of this method should only happen after the
* previously returned future has been completed.
*/
- CompletionStage<Void> acquire();
+ default CompletionStage<Void> acquire() {
+ return acquire(1);
+ }
+
+ /**
+ * Returns a future that is completed once another event would not exceed the rate limit. For
+ * correct functioning, the next invocation of this method should only happen after the
+ * previously returned future has been completed.
+ *
+ * @param requestSize The size of requests.
+ */
+ CompletionStage<Void> acquire(int requestSize);
Review Comment:
This method is the core method of the interface. Other methods, such as `notifyCheckpointComplete` and `notifyStatusChange`, are typically not required to be implemented in most scenarios.
A no-op implementation is not particularly critical because we either do not use the interface at all or require a non-empty implementation solely to enable rate-limiting capabilities.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2478553153
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
Agree to use `numberOf` instead of `size`.
Note that the original comment for the `acquire()` method states:
https://github.com/apache/flink/blob/7b0a8827ffc111fcdd714682fb1c66be9d5d981d/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java#L33
Using `event` is more appropriate than `request` in this context, hence the term `numberOfEvents` is used.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2454001767
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiterStrategy.java:
##########
@@ -27,14 +27,14 @@
* A factory for {@link RateLimiter RateLimiters} which apply rate-limiting to a source sub-task.
*/
@Experimental
-public interface RateLimiterStrategy extends Serializable {
+public interface RateLimiterStrategy<S> extends Serializable {
/**
* Creates a {@link RateLimiter} that lets records through with rate proportional to the
Review Comment:
Addressed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2472407604
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java:
##########
@@ -27,14 +27,25 @@
/** The interface to rate limit execution of methods. */
@NotThreadSafe
@Experimental
-public interface RateLimiter {
+public interface RateLimiter<S> {
/**
* Returns a future that is completed once another event would not exceed the rate limit. For
* correct functioning, the next invocation of this method should only happen after the
* previously returned future has been completed.
*/
- CompletionStage<Void> acquire();
+ default CompletionStage<Void> acquire() {
+ return acquire(1);
+ }
+
+ /**
+ * Returns a future that is completed once another event would not exceed the rate limit. For
+ * correct functioning, the next invocation of this method should only happen after the
+ * previously returned future has been completed.
+ *
+ * @param requestSize The size of requests.
+ */
+ CompletionStage<Void> acquire(int requestSize);
Review Comment:
is it worth setting the default as the no-op implementation
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3469755590
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2472060052
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java:
##########
@@ -156,14 +177,29 @@ public InputStatus pollNext(ReaderOutput<T> output) throws Exception {
// we need to loop here, because we may have to go across splits
while (true) {
+ // Check if the previous record count reached the limit of rateLimiter.
+ if (rateLimitPermissionFuture != null && !rateLimitPermissionFuture.isDone()) {
Review Comment:
In current implementation, we need to check here for **each record** even rateLimitPermissionFuture is null , I believe this will introduce a performance regression. We can avoid this regression as we've known that `rateLimiterStrategy` nullability in constructor.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3465797463
Subtask for document update: https://issues.apache.org/jira/browse/FLINK-38589.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2477568163
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
+1 to use `numberOfRequests` @lvyanquan
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2478553153
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
Agree to use `numberOf` instead of `size`.
Note that the original comment for the `acquire()` method states:
```
Returns a future that is completed once another event would not exceed the rate limit.
```
Using `event` is more appropriate than `request` in this context, hence the term `numberOfEvents` is used.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2472435068
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceReaderBaseTest.java:
##########
@@ -133,18 +140,138 @@ public void close() {}
void testRecordsWithSplitsNotRecycledWhenRecordsLeft() throws Exception {
final TestingRecordsWithSplitIds<String> records =
new TestingRecordsWithSplitIds<>("test-split", "value1", "value2");
- final SourceReader<?, ?> reader = createReaderAndAwaitAvailable("test-split", records);
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Collections.singletonList("test-split"),
+ Collections.singletonList(records),
+ RateLimiterStrategy.noOp());
reader.pollNext(new TestingReaderOutput<>());
assertThat(records.isRecycled()).isFalse();
}
+ @Test
+ void testLimitingRateInSplitReader() throws Exception {
+ String[] recordArr = new String[60];
+ for (int i = 0; i < recordArr.length; i++) {
+ recordArr[i] = "value" + i;
+ }
+ final TestingRecordsWithSplitIds<String> records =
+ new TestingRecordsWithSplitIds<>("test-split", recordArr);
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Collections.singletonList("test-split"),
+ Collections.singletonList(records),
+ RateLimiterStrategy.perSecond(2));
+ TestingReaderOutput testingReaderOutput = new TestingReaderOutput<>();
+ long startTime = System.currentTimeMillis();
+ while (testingReaderOutput.getEmittedRecords().size() < recordArr.length) {
+ reader.pollNext(testingReaderOutput);
+ }
+ // The first few seconds require preheating, there may be a deviation of a few seconds.
+ assertThat(System.currentTimeMillis() - startTime)
+ .isGreaterThan(Duration.ofSeconds(25).toMillis());
+ assertThat(System.currentTimeMillis() - startTime)
+ .isLessThan(Duration.ofSeconds(35).toMillis());
+ }
+
+ @Test
+ void testLimitingRatePerCheckpointInSplitReader() throws Exception {
+ String[] recordArr = new String[30];
+ for (int i = 0; i < recordArr.length; i++) {
+ recordArr[i] = "value" + i;
+ }
+ final TestingRecordsWithSplitIds<String> records =
+ new TestingRecordsWithSplitIds<>("test-split", recordArr);
+ int recordsPerCheckpoint = 2;
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Collections.singletonList("test-split"),
+ Collections.singletonList(records),
+ RateLimiterStrategy.perCheckpoint(recordsPerCheckpoint));
+ TestingReaderOutput testingReaderOutput = new TestingReaderOutput<>();
+ for (int i = 1; i <= recordArr.length / recordsPerCheckpoint; i++) {
+ long startTime = System.currentTimeMillis();
+ while (System.currentTimeMillis() - startTime < Duration.ofSeconds(2).toMillis()) {
+ reader.pollNext(testingReaderOutput);
+ }
+ assertThat(testingReaderOutput.getEmittedRecords().size())
+ .isGreaterThanOrEqualTo(1 + recordsPerCheckpoint * (i - 1));
+ assertThat(testingReaderOutput.getEmittedRecords().size())
+ .isLessThanOrEqualTo(1 + recordsPerCheckpoint * i);
+ reader.notifyCheckpointComplete(i);
+ }
+ }
+
+ @Test
+ void testLimitingRateWithStatusChangeInSplitReader() throws Exception {
+ String[] recordArr = new String[60];
+ for (int i = 0; i < recordArr.length; i++) {
+ recordArr[i] = "value" + i;
+ }
+ final TestingRecordsWithSplitIds<String> firstRecords =
+ new TestingRecordsWithSplitIds<>("test-split1", recordArr);
+ final TestingRecordsWithSplitIds<String> secondRecords =
+ new TestingRecordsWithSplitIds<>("test-split2", recordArr);
+ int maxPerSecond = 2;
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Arrays.asList("test-split1", "test-split2"),
+ Arrays.asList(firstRecords, secondRecords),
+ parallelism ->
+ new SplitAwaredRateLimiter((double) maxPerSecond / parallelism));
+ TestingReaderOutput testingReaderOutput = new TestingReaderOutput<>();
+ long startTime = System.currentTimeMillis();
+ while (testingReaderOutput.getEmittedRecords().size() < 2 * recordArr.length) {
+ reader.pollNext(testingReaderOutput);
+ }
+ // The first few seconds require preheating, there may be a deviation of a few seconds.
+ assertThat(System.currentTimeMillis() - startTime)
+ .isGreaterThan(Duration.ofSeconds(85).toMillis());
+ // The first few seconds require preheating, there may be a deviation of a few seconds.
+ assertThat(System.currentTimeMillis() - startTime)
+ .isLessThan(Duration.ofSeconds(95).toMillis());
+ }
+
+ /** A rate limiter that reduce the maxPerSecond for specific splits. */
+ private static class SplitAwaredRateLimiter
+ implements org.apache.flink.api.connector.source.util.ratelimit.RateLimiter<
+ org.apache.flink.connector.base.source.reader.mocks.TestingSourceSplit> {
+
+ private final Executor limiter =
+ Executors.newSingleThreadExecutor(new ExecutorThreadFactory("flink-rate-limiter"));
+ private RateLimiter rateLimiter;
+ private final double maxPerSecond;
+
+ public SplitAwaredRateLimiter(double maxPerSecond) {
+ this.maxPerSecond = maxPerSecond;
+ this.rateLimiter = RateLimiter.create(maxPerSecond);
+ }
+
+ @Override
+ public CompletionStage<Void> acquire(int requestSize) {
+ return CompletableFuture.runAsync(() -> rateLimiter.acquire(requestSize), limiter);
+ }
+
+ @Override
+ public void notifyStatusChange(
+ org.apache.flink.connector.base.source.reader.mocks.TestingSourceSplit split) {
+ if (!split.splitId().equals("test-split1")) {
+ this.rateLimiter = RateLimiter.create(maxPerSecond / 2);
+ }
+ }
+ }
+
@Test
void testRecordsWithSplitsRecycledWhenEmpty() throws Exception {
final TestingRecordsWithSplitIds<String> records =
new TestingRecordsWithSplitIds<>("test-split", "value1", "value2");
- final SourceReader<?, ?> reader = createReaderAndAwaitAvailable("test-split", records);
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Collections.singletonList("test-split"),
+ Collections.singletonList(records),
+ RateLimiterStrategy.noOp());
Review Comment:
curious about the line below - poll thrice, why do we need to poll 3 times?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3461898534
Thanks for suggestion of @davidradl and @leonardBang, updated.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2476735480
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiterStrategy.java:
##########
@@ -27,14 +27,14 @@
* A factory for {@link RateLimiter RateLimiters} which apply rate-limiting to a source sub-task.
*/
@Experimental
-public interface RateLimiterStrategy extends Serializable {
+public interface RateLimiterStrategy<S> extends Serializable {
Review Comment:
Please add java doc for `<S>`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3435280495
Hi @leonardBang @ruanhang1993, maybe you can help to review this.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #27134:
URL: https://github.com/apache/flink/pull/27134
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3466376107
Rebase master to fix CI failure of StreamExecutionEnvironmentTests::test_set_requirements_with_cached_directory.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2476750158
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java:
##########
@@ -27,14 +27,25 @@
/** The interface to rate limit execution of methods. */
@NotThreadSafe
@Experimental
-public interface RateLimiter {
+public interface RateLimiter<S> {
Review Comment:
I saw the definition `private final RateLimiter<SplitT> rateLimiter;` in SourceReaderBase, what's meaning of generic type `<S>`? and this is a breaking change from `RateLimiter` to `RateLimiter<S>`, I think we should follow the FLIP design.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2473357903
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java:
##########
@@ -27,14 +27,25 @@
/** The interface to rate limit execution of methods. */
@NotThreadSafe
@Experimental
-public interface RateLimiter {
+public interface RateLimiter<S> {
/**
* Returns a future that is completed once another event would not exceed the rate limit. For
* correct functioning, the next invocation of this method should only happen after the
* previously returned future has been completed.
*/
- CompletionStage<Void> acquire();
+ default CompletionStage<Void> acquire() {
+ return acquire(1);
+ }
+
+ /**
+ * Returns a future that is completed once another event would not exceed the rate limit. For
+ * correct functioning, the next invocation of this method should only happen after the
+ * previously returned future has been completed.
+ *
+ * @param requestSize The size of requests.
+ */
+ CompletionStage<Void> acquire(int requestSize);
Review Comment:
This method is the core method of the interface. Other methods, such as notifyCheckpointComplete and notifyStatusChange, are typically not required to be implemented in most scenarios.
A no-op implementation is not particularly critical because we either do not use the interface at all or require a non-empty implementation solely to enable rate-limiting capabilities.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2477568163
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
+ to use `numberOfRequests` @lvyanquan
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3465779665
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2473417731
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceReaderBaseTest.java:
##########
@@ -133,18 +140,138 @@ public void close() {}
void testRecordsWithSplitsNotRecycledWhenRecordsLeft() throws Exception {
final TestingRecordsWithSplitIds<String> records =
new TestingRecordsWithSplitIds<>("test-split", "value1", "value2");
- final SourceReader<?, ?> reader = createReaderAndAwaitAvailable("test-split", records);
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Collections.singletonList("test-split"),
+ Collections.singletonList(records),
+ RateLimiterStrategy.noOp());
reader.pollNext(new TestingReaderOutput<>());
assertThat(records.isRecycled()).isFalse();
}
+ @Test
+ void testLimitingRateInSplitReader() throws Exception {
+ String[] recordArr = new String[60];
+ for (int i = 0; i < recordArr.length; i++) {
+ recordArr[i] = "value" + i;
+ }
+ final TestingRecordsWithSplitIds<String> records =
+ new TestingRecordsWithSplitIds<>("test-split", recordArr);
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Collections.singletonList("test-split"),
+ Collections.singletonList(records),
+ RateLimiterStrategy.perSecond(2));
+ TestingReaderOutput testingReaderOutput = new TestingReaderOutput<>();
+ long startTime = System.currentTimeMillis();
+ while (testingReaderOutput.getEmittedRecords().size() < recordArr.length) {
+ reader.pollNext(testingReaderOutput);
+ }
+ // The first few seconds require preheating, there may be a deviation of a few seconds.
+ assertThat(System.currentTimeMillis() - startTime)
+ .isGreaterThan(Duration.ofSeconds(25).toMillis());
+ assertThat(System.currentTimeMillis() - startTime)
+ .isLessThan(Duration.ofSeconds(35).toMillis());
+ }
+
+ @Test
+ void testLimitingRatePerCheckpointInSplitReader() throws Exception {
+ String[] recordArr = new String[30];
+ for (int i = 0; i < recordArr.length; i++) {
+ recordArr[i] = "value" + i;
+ }
+ final TestingRecordsWithSplitIds<String> records =
+ new TestingRecordsWithSplitIds<>("test-split", recordArr);
+ int recordsPerCheckpoint = 2;
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Collections.singletonList("test-split"),
+ Collections.singletonList(records),
+ RateLimiterStrategy.perCheckpoint(recordsPerCheckpoint));
+ TestingReaderOutput testingReaderOutput = new TestingReaderOutput<>();
+ for (int i = 1; i <= recordArr.length / recordsPerCheckpoint; i++) {
+ long startTime = System.currentTimeMillis();
+ while (System.currentTimeMillis() - startTime < Duration.ofSeconds(2).toMillis()) {
+ reader.pollNext(testingReaderOutput);
+ }
+ assertThat(testingReaderOutput.getEmittedRecords().size())
+ .isGreaterThanOrEqualTo(1 + recordsPerCheckpoint * (i - 1));
+ assertThat(testingReaderOutput.getEmittedRecords().size())
+ .isLessThanOrEqualTo(1 + recordsPerCheckpoint * i);
+ reader.notifyCheckpointComplete(i);
+ }
+ }
+
+ @Test
+ void testLimitingRateWithStatusChangeInSplitReader() throws Exception {
+ String[] recordArr = new String[60];
+ for (int i = 0; i < recordArr.length; i++) {
+ recordArr[i] = "value" + i;
+ }
+ final TestingRecordsWithSplitIds<String> firstRecords =
+ new TestingRecordsWithSplitIds<>("test-split1", recordArr);
+ final TestingRecordsWithSplitIds<String> secondRecords =
+ new TestingRecordsWithSplitIds<>("test-split2", recordArr);
+ int maxPerSecond = 2;
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Arrays.asList("test-split1", "test-split2"),
+ Arrays.asList(firstRecords, secondRecords),
+ parallelism ->
+ new SplitAwaredRateLimiter((double) maxPerSecond / parallelism));
+ TestingReaderOutput testingReaderOutput = new TestingReaderOutput<>();
+ long startTime = System.currentTimeMillis();
+ while (testingReaderOutput.getEmittedRecords().size() < 2 * recordArr.length) {
+ reader.pollNext(testingReaderOutput);
+ }
+ // The first few seconds require preheating, there may be a deviation of a few seconds.
+ assertThat(System.currentTimeMillis() - startTime)
+ .isGreaterThan(Duration.ofSeconds(85).toMillis());
+ // The first few seconds require preheating, there may be a deviation of a few seconds.
+ assertThat(System.currentTimeMillis() - startTime)
+ .isLessThan(Duration.ofSeconds(95).toMillis());
+ }
+
+ /** A rate limiter that reduce the maxPerSecond for specific splits. */
+ private static class SplitAwaredRateLimiter
+ implements org.apache.flink.api.connector.source.util.ratelimit.RateLimiter<
+ org.apache.flink.connector.base.source.reader.mocks.TestingSourceSplit> {
+
+ private final Executor limiter =
+ Executors.newSingleThreadExecutor(new ExecutorThreadFactory("flink-rate-limiter"));
+ private RateLimiter rateLimiter;
+ private final double maxPerSecond;
+
+ public SplitAwaredRateLimiter(double maxPerSecond) {
+ this.maxPerSecond = maxPerSecond;
+ this.rateLimiter = RateLimiter.create(maxPerSecond);
+ }
+
+ @Override
+ public CompletionStage<Void> acquire(int requestSize) {
+ return CompletableFuture.runAsync(() -> rateLimiter.acquire(requestSize), limiter);
+ }
+
+ @Override
+ public void notifyStatusChange(
+ org.apache.flink.connector.base.source.reader.mocks.TestingSourceSplit split) {
+ if (!split.splitId().equals("test-split1")) {
+ this.rateLimiter = RateLimiter.create(maxPerSecond / 2);
+ }
+ }
+ }
+
@Test
void testRecordsWithSplitsRecycledWhenEmpty() throws Exception {
final TestingRecordsWithSplitIds<String> records =
new TestingRecordsWithSplitIds<>("test-split", "value1", "value2");
- final SourceReader<?, ?> reader = createReaderAndAwaitAvailable("test-split", records);
+ final SourceReader<?, ?> reader =
+ createReaderAndAwaitAvailable(
+ Collections.singletonList("test-split"),
+ Collections.singletonList(records),
+ RateLimiterStrategy.noOp());
Review Comment:
This test method has existed previously and is not newly added.
The first two `reader.pollNext()` operations are designed to exhaust all data from the split, while the third `reader.pollNext()` is explicitly used to trigger the `SourceReaderBase#finishCurrentFetch()` method, which in turn invokes `RecordsWithSplitIds#recycle()`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2472417185
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
When we say requestSize - does this mean the size of requests. Looking at the code it seems to be more like the number of requests, if so I would suggest renaming this numberOfRequests to make it clearer.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2477551540
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
Ok thanks for the clarification. numberOfRequests seems better to me as it is not todo the with the size of individual requests. It is up to you if you want to change or not.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3426516514
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a7fc3464b004c323db8d7e98a459ddc036397d05",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a7fc3464b004c323db8d7e98a459ddc036397d05",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a7fc3464b004c323db8d7e98a459ddc036397d05 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2471992520
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBase.java:
##########
@@ -81,6 +82,27 @@ public SingleThreadMultiplexSourceReaderBase(
context);
}
+ /**
+ * The primary constructor for the source reader.
Review Comment:
`primary constructor` should only have one, but current we have two. And I think you need to update the method java notes.
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java:
##########
@@ -435,4 +493,72 @@ private boolean isEndOfStreamReached(T record) {
return isStreamEnd;
}
}
+
+ /**
+ * A wrapper around {@link SourceOutput} that counts the number of records emitted.
+ *
+ * <p>This wrapper is used when rate limiting is enabled to track how many records have been
+ * emitted since the last rate limit check, allowing the reader to properly apply backpressure
+ * when the rate limit is exceeded.
+ *
+ * @param <T> The type of records being emitted
+ */
+ private static final class RecordCountingSourceOutputWrapper<T> implements SourceOutput<T> {
+ /** The underlying source output to delegate to. */
+ final SourceOutput<T> sourceOutput;
+
+ /** The number of records emitted since the last reset. */
+ int recordCount;
Review Comment:
`last reset` is no proper here, window is better, how about
```
/** Count of records handled during the current rate-limiting window. */
int currentWindowRecordCount;
```
?
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBase.java:
##########
@@ -108,4 +130,26 @@ public SingleThreadMultiplexSourceReaderBase(
SourceReaderContext context) {
super(splitFetcherManager, recordEmitter, eofRecordEvaluator, config, context);
}
+
+ /**
+ * This constructor behaves like {@link #SingleThreadMultiplexSourceReaderBase(Supplier,
+ * RecordEmitter, Configuration, SourceReaderContext)}, but accepts a specific {@link
Review Comment:
we can update this java doc with an end `but accepts a specific {@link RateLimiterStrategy}.`
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java:
##########
@@ -156,14 +177,29 @@ public InputStatus pollNext(ReaderOutput<T> output) throws Exception {
// we need to loop here, because we may have to go across splits
while (true) {
+ // Check if the previous record count reached the limit of rateLimiter.
+ if (rateLimitPermissionFuture != null && !rateLimitPermissionFuture.isDone()) {
Review Comment:
In current implementation, we need to check here for **each record `** even rateLimitPermissionFuture is null , I believe this will introduce a performance regression. We can avoid this regression as we've known that `rateLimiterStrategy` nullability in constructor.
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java:
##########
@@ -435,4 +493,72 @@ private boolean isEndOfStreamReached(T record) {
return isStreamEnd;
}
}
+
+ /**
+ * A wrapper around {@link SourceOutput} that counts the number of records emitted.
+ *
+ * <p>This wrapper is used when rate limiting is enabled to track how many records have been
+ * emitted since the last rate limit check, allowing the reader to properly apply backpressure
+ * when the rate limit is exceeded.
+ *
+ * @param <T> The type of records being emitted
+ */
+ private static final class RecordCountingSourceOutputWrapper<T> implements SourceOutput<T> {
Review Comment:
We can rename RecordCountingSourceOutputWrapper => RateLimitingSourceOutputWrapper, which is more Intuitively？
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3466958250
Updated based on comments.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3465856289
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2448500761
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiterStrategy.java:
##########
@@ -27,14 +27,14 @@
* A factory for {@link RateLimiter RateLimiters} which apply rate-limiting to a source sub-task.
*/
@Experimental
-public interface RateLimiterStrategy extends Serializable {
+public interface RateLimiterStrategy<S> extends Serializable {
/**
* Creates a {@link RateLimiter} that lets records through with rate proportional to the
Review Comment:
nit: don't we mean inversely proportional to the parallelism? Though this is a bit mathematical. Maybe
` Creates a {@link RateLimiter} that limits the rate of records going through. When there is parallelism, the limiting rate is evenly reduced per subtask, such that all the sub-tasks limiting rates equals the cumulative limitting rate. `
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3556674394
@lvyanquan can we also markt the `RateLimitingStrategy` `PublicEvolving` as part of this change? It's kind of strange that now a `PublicEvolving` component like the source relies on a `Experimental` interface.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2472417185
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
When we say requestSize - does this mean the size of requests? Looking at the code it seems to relate to the number of futures we want to acquire, if so I would suggest renaming this. Maybe numberOfFuturesToAcquire.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2476731768
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java:
##########
@@ -435,4 +500,73 @@ private boolean isEndOfStreamReached(T record) {
return isStreamEnd;
}
}
+
+ /**
+ * A wrapper around {@link SourceOutput} that counts the number of records during the current
+ * rate-limiting window.
+ *
+ * <p>This wrapper is used when rate limiting is enabled to track how many records have been
+ * emitted since the last rate limit check, allowing the reader to properly apply backpressure
+ * when the rate limit is exceeded.
+ *
+ * @param <T> The type of records being emitted
+ */
+ private static final class RateLimitingSourceOutputWrapper<T> implements SourceOutput<T> {
+ /** The underlying source output to delegate to. */
+ final SourceOutput<T> sourceOutput;
+
+ /** Count of records handled during the current rate-limiting window. */
+ int currentWindowRecordCount;
Review Comment:
Could be private ?
```suggestion
private int currentWindowRecordCount;
```
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java:
##########
@@ -27,14 +27,25 @@
/** The interface to rate limit execution of methods. */
@NotThreadSafe
@Experimental
-public interface RateLimiter {
+public interface RateLimiter<S> {
Review Comment:
I saw the definition `private final RateLimiter<SplitT> rateLimiter;` in SourceReaderBase, what's meaning of generic type `<S>`? and this is a breaking change from `RateLimiter` to `RateLimiter<S>`.
##########
flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiterStrategy.java:
##########
@@ -27,14 +27,14 @@
* A factory for {@link RateLimiter RateLimiters} which apply rate-limiting to a source sub-task.
*/
@Experimental
-public interface RateLimiterStrategy extends Serializable {
+public interface RateLimiterStrategy<S> extends Serializable {
Review Comment:
Please add java doc for `<S>`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2478553153
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
Agree to use `numberOf` instead of `size`.
Note that the original comment for the `acquire()` method states:
https://github.com/apache/flink/blob/7b0a8827ffc111fcdd714682fb1c66be9d5d981d/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java#L33
Using `event` is more appropriate than `request` in this context, hence the term `numberOfEvents` is used now.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2472721740
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBase.java:
##########
@@ -81,6 +82,27 @@ public SingleThreadMultiplexSourceReaderBase(
context);
}
+ /**
+ * The primary constructor for the source reader.
Review Comment:
Agreed - I am not sure how useful the term is. Should be used for minimal parameters or the maximum parameters.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #27134:
URL: https://github.com/apache/flink/pull/27134#discussion_r2473477618
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java:
##########
@@ -374,19 +374,19 @@ private void assertUnalignedCheckpointInNonSink(StreamGraph streamGraph) {
IntegerTypeInfo.INT_TYPE_INFO);
}
- private static class BurstingRateLimiter implements RateLimiter {
- private final RateLimiter rateLimiter;
+ private static class BurstingRateLimiter implements RateLimiter<Void> {
+ private final RateLimiter<Void> rateLimiter;
private final int numCheckpointCooldown;
private int cooldown;
public BurstingRateLimiter(int recordPerCycle, int numCheckpointCooldown) {
- rateLimiter = new GatedRateLimiter(recordPerCycle);
+ rateLimiter = new GatedRateLimiter<>(recordPerCycle);
this.numCheckpointCooldown = numCheckpointCooldown;
}
@Override
- public CompletionStage<Void> acquire() {
- CompletionStage<Void> stage = rateLimiter.acquire();
+ public CompletionStage<Void> acquire(int requestSize) {
Review Comment:
The returned `Future` here represents a single asynchronous execution result, and there is no scenario where multiple `Future` instances are returned. Therefore, the variable name `numberOfFutures` is not appropriate.
The input parameter here represents the number of requests, which could specifically refer to the number (or even size) of data records. Since this interface is not exclusively called by `SourceReader`, I used the more generic term "Request" to abstract the concept beyond data-specific contexts.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source. [flink]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #27134:
URL: https://github.com/apache/flink/pull/27134#issuecomment-3587654288
> @lvyanquan can we also markt the `RateLimitingStrategy` `PublicEvolving` as part of this change? It's kind of strange that now a `PublicEvolving` component like the source relies on a `Experimental` interface.
Thank you for pointing this out. I agree with your perspective, and I will submit a PR to modify this annotation.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org