(seatunnel) branch dev updated: [Fix][Core] Do not charge the byte rate limiter for a zero-byte row (#11766)

[github-bot]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new 5dc71549ad [Fix][Core] Do not charge the byte rate limiter for a 
zero-byte row (#11766)
5dc71549ad is described below

commit 5dc71549adb79a798f4d8a9de64523830d57de39
Author: ZIHAN DAI <99155080+pd...@users.noreply.github.com>
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();