This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 8ec1b0b9828 [FLINK-40326][core] Do not over-grant capacity in
GatedRateLimiter
8ec1b0b9828 is described below
commit 8ec1b0b9828f1d4caf66885005dd7c1ca5b97cc9
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Mon Aug 3 18:44:08 2026 +0200
[FLINK-40326][core] Do not over-grant capacity in GatedRateLimiter
The decrement in acquire() was generalized from a constant 1 to
numberOfEvents as part of FLIP-535, but the gate was left closing only at
capacityLeft <= 0, so a cycle can admit more events than capacityPerCycle.
The gate is now closed unless the capacity left covers the whole request.
---
.../source/util/ratelimit/GatedRateLimiter.java | 16 ++---
.../source/util/ratelimit/RateLimiter.java | 2 +-
.../source/lib/util/GatedRateLimiterTest.java | 76 +++++++++++++++++++++-
3 files changed, 84 insertions(+), 10 deletions(-)
diff --git
a/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/GatedRateLimiter.java
b/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/GatedRateLimiter.java
index 2fa991af013..75fb0b9303a 100644
---
a/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/GatedRateLimiter.java
+++
b/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/GatedRateLimiter.java
@@ -28,8 +28,9 @@ import static
org.apache.flink.util.Preconditions.checkArgument;
/**
* An implementation of {@link RateLimiter} that completes defined number of
futures in-between the
- * external notification events. The first cycle completes immediately,
without waiting for the
- * external notifications.
+ * external notification events. The first cycle does not wait for an external
notification: its
+ * capacity is available from the start, so requests complete immediately for
as long as the
+ * capacity left covers them.
*/
@Internal
public class GatedRateLimiter<Split extends SourceSplit> implements
RateLimiter<Split> {
@@ -37,6 +38,9 @@ public class GatedRateLimiter<Split extends SourceSplit>
implements RateLimiter<
private final int capacityPerCycle;
private int capacityLeft;
+ /** Completed while the current cycle has capacity left, incomplete once
it has run out. */
+ private CompletableFuture<Void> gatingFuture =
CompletableFuture.completedFuture(null);
+
/**
* Instantiates a new GatedRateLimiter.
*
@@ -48,14 +52,10 @@ public class GatedRateLimiter<Split extends SourceSplit>
implements RateLimiter<
this.capacityLeft = capacityPerCycle;
}
- transient CompletableFuture<Void> gatingFuture = null;
-
@Override
public CompletionStage<Void> acquire(int numberOfEvents) {
- if (gatingFuture == null) {
- gatingFuture = CompletableFuture.completedFuture(null);
- }
- if (capacityLeft <= 0) {
+ checkArgument(numberOfEvents > 0, "Number of events has to be a
positive number.");
+ if (capacityLeft < numberOfEvents) {
gatingFuture = new CompletableFuture<>();
}
return gatingFuture.thenRun(() -> capacityLeft -= numberOfEvents);
diff --git
a/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java
b/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java
index 06de7bc7269..bc2a14ab191 100644
---
a/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java
+++
b/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java
@@ -48,7 +48,7 @@ public interface RateLimiter<SplitT extends SourceSplit> {
* correct functioning, the next invocation of this method should only
happen after the
* previously returned future has been completed.
*
- * @param numberOfEvents The number of events.
+ * @param numberOfEvents The number of events, which has to be a positive
number.
*/
CompletionStage<Void> acquire(int numberOfEvents);
diff --git
a/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/util/GatedRateLimiterTest.java
b/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/util/GatedRateLimiterTest.java
index 15a034e0f7d..21c905a6374 100644
---
a/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/util/GatedRateLimiterTest.java
+++
b/flink-tests/src/test/java/org/apache/flink/api/connector/source/lib/util/GatedRateLimiterTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.connector.source.lib.util;
+import
org.apache.flink.api.connector.source.lib.NumberSequenceSource.NumberSequenceSplit;
import org.apache.flink.api.connector.source.util.ratelimit.GatedRateLimiter;
import org.junit.jupiter.api.Test;
@@ -25,6 +26,7 @@ import org.junit.jupiter.api.Test;
import java.util.concurrent.CompletionStage;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
class GatedRateLimiterTest {
@@ -32,7 +34,8 @@ class GatedRateLimiterTest {
void testCapacityNotExceededOnCheckpoint() {
int capacityPerCycle = 5;
- final GatedRateLimiter gatedRateLimiter = new
GatedRateLimiter(capacityPerCycle);
+ final GatedRateLimiter<NumberSequenceSplit> gatedRateLimiter =
+ new GatedRateLimiter<>(capacityPerCycle);
for (int x = 0; x < capacityPerCycle; x++) {
assertThat(gatedRateLimiter.acquire()).isCompleted();
}
@@ -50,4 +53,75 @@ class GatedRateLimiterTest {
CompletionStage<Void> postCheckpoint = gatedRateLimiter.acquire();
assertThat(postCheckpoint).isNotCompleted();
}
+
+ @Test
+ void testCapacityNotExceededWhenAcquiringMultipleEvents() {
+ int capacityPerCycle = 5;
+
+ final GatedRateLimiter<NumberSequenceSplit> gatedRateLimiter =
+ new GatedRateLimiter<>(capacityPerCycle);
+ assertThat(gatedRateLimiter.acquire(3)).isCompleted();
+
+ // Only two permits are left in this cycle, so a request for three
events has to wait even
+ // though the remaining capacity is still greater than zero.
+ CompletionStage<Void> exceedsRemainingCapacity =
gatedRateLimiter.acquire(3);
+ assertThat(exceedsRemainingCapacity).isNotCompleted();
+
+ gatedRateLimiter.notifyCheckpointComplete(0);
+
+ assertThat(exceedsRemainingCapacity).isCompleted();
+ }
+
+ @Test
+ void testRequestLargerThanCapacityIsReleasedByNextCycle() {
+ final GatedRateLimiter<NumberSequenceSplit> gatedRateLimiter = new
GatedRateLimiter<>(2);
+
+ // A single request may legitimately exceed the capacity of an entire
cycle. It must not
+ // deadlock: resetting the capacity on the next completed checkpoint
releases it.
+ CompletionStage<Void> exceedsWholeCycle = gatedRateLimiter.acquire(3);
+ assertThat(exceedsWholeCycle).isNotCompleted();
+
+ gatedRateLimiter.notifyCheckpointComplete(0);
+
+ assertThat(exceedsWholeCycle).isCompleted();
+
+ // Because completing it took 3 events from a cycle that only had 2, a
further checkpoint is
+ // needed before requests are allowed again.
+ CompletionStage<Void> followingRequest = gatedRateLimiter.acquire(1);
+ assertThat(followingRequest).isNotCompleted();
+
+ gatedRateLimiter.notifyCheckpointComplete(1);
+
+ assertThat(followingRequest).isCompleted();
+ assertThat(gatedRateLimiter.acquire(1)).isCompleted();
+ }
+
+ @Test
+ void testCheckpointCompleteBeforeFirstAcquire() {
+ int capacityPerCycle = 5;
+
+ final GatedRateLimiter<NumberSequenceSplit> gatedRateLimiter =
+ new GatedRateLimiter<>(capacityPerCycle);
+
+ // A checkpoint can complete before the reader has emitted anything,
for instance while it
+ // is still waiting for its first split assignment.
+ gatedRateLimiter.notifyCheckpointComplete(0);
+
+ for (int x = 0; x < capacityPerCycle; x++) {
+ assertThat(gatedRateLimiter.acquire()).isCompleted();
+ }
+ assertThat(gatedRateLimiter.acquire()).isNotCompleted();
+ }
+
+ @Test
+ void testNonPositiveNumberOfEventsIsRejected() {
+ final GatedRateLimiter<NumberSequenceSplit> gatedRateLimiter = new
GatedRateLimiter<>(5);
+
+ assertThatThrownBy(() -> gatedRateLimiter.acquire(0))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("positive");
+ assertThatThrownBy(() -> gatedRateLimiter.acquire(-1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("positive");
+ }
}
