(fluss) branch main updated: [client] release pending fetch semaphore on discarding (#3043)

[aborisov]

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

fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 570f09be6 [client] release pending fetch semaphore on discarding 
(#3043)
570f09be6 is described below

commit 570f09be6aa8a9317d23cb4c40c2865aac71a4e4
Author: Junfan Zhang <zus...@apache.org>
AuthorDate: Fri May 8 17:56:07 2026 +0800

    [client] release pending fetch semaphore on discarding (#3043)
    
    * [client] release semaphore when discarding completedFetch
    
    * add test
    
    * spotless
    
    ---------
    
    Co-authored-by: Junfan Zhang <zhangjun...@qiyi.com>
---
 .../client/table/scanner/log/LogFetchCollector.java     |  6 +++++-
 .../client/table/scanner/log/LogFetchCollectorTest.java | 17 +++++++++++++++++
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git 
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
 
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
index 5961de0ec..05c43eca0 100644
--- 
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
+++ 
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
@@ -98,7 +98,11 @@ public class LogFetchCollector {
 
                     if (!completedFetch.isInitialized()) {
                         try {
-                            
logFetchBuffer.setNextInLineFetch(initialize(completedFetch));
+                            CompletedFetch initialized = 
initialize(completedFetch);
+                            logFetchBuffer.setNextInLineFetch(initialized);
+                            if (initialized == null) {
+                                completedFetch.drain();
+                            }
                         } catch (Exception e) {
                             // Remove a completedFetch upon a parse with 
exception if
                             // (1) it contains no records, and
diff --git 
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
 
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
index 08b68563c..e769ebf4e 100644
--- 
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
+++ 
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
@@ -266,6 +266,23 @@ public class LogFetchCollectorTest {
                 tableBucket, resultForBucket, readContext, logScannerStatus, 
true, offset, null);
     }
 
+    @Test
+    void testCollectDrainsDiscardedFetch() throws Exception {
+        TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+        CompletedFetch completedFetch =
+                makeCompletedFetch(
+                        tb,
+                        new FetchLogResultForBucket(tb, 
genMemoryLogRecordsByObject(DATA1), 10L),
+                        0L);
+        logFetchBuffer.add(completedFetch);
+        logScannerStatus.unassignScanBuckets(Collections.singletonList(tb));
+
+        ScanRecords records = logFetchCollector.collectFetch(logFetchBuffer);
+
+        assertThat(records.buckets()).isEmpty();
+        assertThat(completedFetch.isConsumed()).isTrue();
+    }
+
     @Test
     void testUpdateBeforeAndAfterNeverSplitAcrossPolls() throws Exception {
         // Create records: INSERT, UPDATE_BEFORE, UPDATE_AFTER, INSERT