This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11766-03ef9f1c0010a60eb29490ceead1309ad825c555 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 5dc71549adb79a798f4d8a9de64523830d57de39 Author: ZIHAN DAI <[email protected]> AuthorDate: Wed Sep 2 11:11:38 2026 +0000 [Fix][Core] Do not charge the byte rate limiter for a zero-byte row (#11766) --- .../core/starter/flowcontrol/FlowControlGate.java | 18 +++++++++++++++++- .../core/starter/flowcontrol/FlowControlGateTest.java | 18 ++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/flowcontrol/FlowControlGate.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/flowcontrol/FlowControlGate.java index fed891a5eb..8cf235abb7 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/flowcontrol/FlowControlGate.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/flowcontrol/FlowControlGate.java @@ -44,10 +44,26 @@ public class FlowControlGate { } public void audit(SeaTunnelRow row) { - bytesRateLimiter.ifPresent(rateLimiter -> rateLimiter.acquire(row.getBytesSize())); + bytesRateLimiter.ifPresent(rateLimiter -> acquireBytes(rateLimiter, row)); countRateLimiter.ifPresent(RateLimiter::acquire); } + /** + * Charge the byte limiter for this row, unless the row measures zero bytes. + * + * <p>A row can legitimately measure zero bytes: {@link SeaTunnelRow#getBytesSize()} counts a + * null field as 0 and a String as its length, so an all-null row, or one of only empty strings, + * sums to 0. {@link RateLimiter#acquire(int)} rejects a non-positive permit count, so charging + * it would abort the task instead of throttling it. Such a row consumes no byte budget; the + * count limiter is what bounds its rate. + */ + private void acquireBytes(RateLimiter rateLimiter, SeaTunnelRow row) { + int bytesSize = row.getBytesSize(); + if (bytesSize > 0) { + rateLimiter.acquire(bytesSize); + } + } + public static FlowControlGate create(FlowControlStrategy flowControlStrategy) { return new FlowControlGate(flowControlStrategy); } diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/flowcontrol/FlowControlGateTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/flowcontrol/FlowControlGateTest.java index 1bcb695bf5..463b07dfbb 100644 --- a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/flowcontrol/FlowControlGateTest.java +++ b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/flowcontrol/FlowControlGateTest.java @@ -33,6 +33,24 @@ public class FlowControlGateTest { private static final int rowSize = 181; + @Test + public void testRowWhoseFieldsAreAllNull() { + FlowControlGate flowControlGate = FlowControlGate.create(FlowControlStrategy.ofBytes(100)); + SeaTunnelRow row = new SeaTunnelRow(new Object[] {null, null}); + + Assertions.assertEquals(0, row.getBytesSize()); + Assertions.assertDoesNotThrow(() -> flowControlGate.audit(row)); + } + + @Test + public void testRowWhoseFieldsAreAllEmptyStrings() { + FlowControlGate flowControlGate = FlowControlGate.create(FlowControlStrategy.ofBytes(100)); + SeaTunnelRow row = new SeaTunnelRow(new Object[] {"", ""}); + + Assertions.assertEquals(0, row.getBytesSize()); + Assertions.assertDoesNotThrow(() -> flowControlGate.audit(row)); + } + @Test public void testWithBytes() { Clock clock = Clock.systemDefaultZone();
