(fluss) branch main updated: [server] Keep replica fetcher count within configured limit (#4508)

[yu...@apache.org]

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

luoyuxia 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 5fe06edf7 [server] Keep replica fetcher count within configured limit (#4508)
5fe06edf7 is described below

commit 5fe06edf73c2ab4b8f0a9a7a5926825a7d5ca0e6
Author: fhan <aa...@gmail.com>
AuthorDate: Tue Sep 29 19:36:15 2026 +0800

    [server] Keep replica fetcher count within configured limit (#4508)
---
 .../replica/fetcher/ReplicaFetcherManager.java     |  5 +++--
 .../replica/fetcher/ReplicaFetcherManagerTest.java | 24 +++++++++++++++++++---
 2 files changed, 24 insertions(+), 5 deletions(-)

diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManager.java
index 9e4479b3e..9030cb2fb 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManager.java
@@ -167,8 +167,9 @@ public class ReplicaFetcherManager {
         }
     }
 
-    private int getFetcherId(TableBucket tableBucket) {
-        return tableBucket.hashCode() % numFetchersPerServer;
+    @VisibleForTesting
+    int getFetcherId(TableBucket tableBucket) {
+        return Math.floorMod(tableBucket.hashCode(), numFetchersPerServer);
     }
 
     private ReplicaFetcherThread addAndStartFetcherThread(
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManagerTest.java
index 5c059dea1..4ecce7434 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManagerTest.java
@@ -74,7 +74,6 @@ class ReplicaFetcherManagerTest extends ReplicaTestBase {
 
     @Test
     void testAddAndRemoveBucket() {
-        int numFetchers = 2;
         ReplicaFetcherManager fetcherManager =
                 new TestingReplicaFetcherManager(TABLET_SERVER_ID, replicaManager, fetcherThread);
 
@@ -111,14 +110,14 @@ class ReplicaFetcherManagerTest extends ReplicaTestBase {
                 fetcherManager.getFetcherThreadMap();
         assertThat(fetcherThreadMap.size()).isEqualTo(1);
         ReplicaFetcherThread thread =
-                fetcherThreadMap.get(new ServerIdAndFetcherId(1, tb.hashCode() % numFetchers));
+                fetcherThreadMap.get(new ServerIdAndFetcherId(1, fetcherManager.getFetcherId(tb)));
         assertThat(thread).isEqualTo(fetcherThread);
         assertThat(fetcherThread.fetchStatus(tb).isPresent()).isTrue();
 
         fetcherManager.removeFetcherForBuckets(Collections.singleton(tb));
         // the fetcher thread will not be moved out from the map.
         assertThat(fetcherThreadMap.size()).isEqualTo(1);
-        thread = fetcherThreadMap.get(new ServerIdAndFetcherId(1, tb.hashCode() % numFetchers));
+        thread = fetcherThreadMap.get(new ServerIdAndFetcherId(1, fetcherManager.getFetcherId(tb)));
 
         assertThat(thread).isEqualTo(fetcherThread);
         assertThat(fetcherThread.fetchStatus(tb).isPresent()).isFalse();
@@ -127,6 +126,25 @@ class ReplicaFetcherManagerTest extends ReplicaTestBase {
         assertThat(fetcherThreadMap.size()).isEqualTo(0);
     }
 
+    @Test
+    void testFetcherIdWithinConfiguredRange() {
+        int numFetchers = 2;
+        conf.set(ConfigOptions.LOG_REPLICA_FETCHER_NUMBER, numFetchers);
+        ReplicaFetcherManager fetcherManager =
+                new ReplicaFetcherManager(
+                        conf, null, TABLET_SERVER_ID, replicaManager, id -> Optional.of(leader));
+
+        TableBucket positiveHashBucket = new TableBucket(0L, 0);
+        TableBucket negativeHashBucket = new TableBucket(Integer.MAX_VALUE, 1);
+        TableBucket evenHashBucket = new TableBucket(0L, 1);
+        assertThat(positiveHashBucket.hashCode()).isPositive();
+        assertThat(negativeHashBucket.hashCode()).isNegative();
+
+        assertThat(fetcherManager.getFetcherId(positiveHashBucket)).isEqualTo(1);
+        assertThat(fetcherManager.getFetcherId(negativeHashBucket)).isEqualTo(1);
+        assertThat(fetcherManager.getFetcherId(evenHashBucket)).isZero();
+    }
+
     @Test
     void testDoesNotAddFetcherWhenFollowerHasNoLeader() {
         TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);