[INLONG-11717][SDK] Add out-of-bounds check when in getClientByRoundR…

…obin() (#11718) * [INLONG-11717][SDK] Add out-of-bounds check when in getClientByRoundRobin() * [INLONG-11717][SDK] Add out-of-bounds check when in getClientByRoundRobin() --------- Co-authored-by: gosonzhang <[email protected]>
apache · Feb 7, 2025 · 17de673 · 17de673
1 parent 2776a98
commit 17de673
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 2 deletions.
diff --git a/...sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/DefClientMgr.java b/...sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/DefClientMgr.java
@@ -58,6 +58,7 @@ public class DefClientMgr implements ConfigHolder {
     private static final Logger logger = LoggerFactory.getLogger(DefClientMgr.class);
     private static final LogCounter logCounter = new LogCounter(10, 100000, 60 * 1000L);
     private static final LogCounter updConExptCnt = new LogCounter(10, 100000, 60 * 1000L);
+    private static final LogCounter indexExptCnt = new LogCounter(10, 100000, 60 * 1000L);
     private static final LogCounter exptCounter = new LogCounter(10, 100000, 60 * 1000L);
     private static final byte[] hbMsgBody = ProxyUtils.getLocalIp().getBytes(StandardCharsets.UTF_8);
 
@@ -197,14 +198,24 @@ public Tuple2<SendResult, NettyClient> getClientByRoundRobin(MutableBoolean allC
         if (curNodeSize == 0) {
             return new Tuple2<>(SendResult.EMPTY_ACTIVE_NODE_SET, null);
         }
+        int indexPos;
         String curNode;
         NettyClient client;
         NettyClient backClient = null;
         int nullClientCnt = 0;
         int incFlightFailCnt = 0;
         int startPos = reqSendIndex.getAndIncrement();
         for (int step = 0; step < curNodeSize; step++) {
-            curNode = curNodes.get(Math.abs(startPos++) % curNodeSize);
+            startPos = Math.abs(startPos + 1);
+            indexPos = startPos % curNodeSize;
+            if (indexPos >= curNodes.size()) {
+                if (indexExptCnt.shouldPrint()) {
+                    logger.warn("IndexOutOfBounds, startPos={}, curNodeSize={}, indexPos={}， realSize={}",
+                            startPos, curNodeSize, indexPos, curNodes.size());
+                }
+                continue;
+            }
+            curNode = curNodes.get(indexPos);
             client = usingClientMaps.get(curNode);
             if (client == null) {
                 nullClientCnt++;

diff --git a/...dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/TcpClientMgr.java b/...dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/TcpClientMgr.java
@@ -67,6 +67,7 @@ public class TcpClientMgr implements ClientMgr {
     private static final Logger logger = LoggerFactory.getLogger(TcpClientMgr.class);
     private static final LogCounter sendExceptCnt = new LogCounter(10, 100000, 60 * 1000L);
     private static final LogCounter updConExptCnt = new LogCounter(10, 100000, 60 * 1000L);
+    private static final LogCounter indexExptCnt = new LogCounter(10, 100000, 60 * 1000L);
     private static final LogCounter exptCounter = new LogCounter(10, 100000, 60 * 1000L);
     private static final LogCounter callbackExceptCnt = new LogCounter(10, 100000, 60 * 1000L);
     private static final AtomicLong timerRefCnt = new AtomicLong(0);
@@ -326,14 +327,24 @@ public boolean getClientByRoundRobin(ProcessResult procResult) {
         if (curNodeSize == 0) {
             return procResult.setFailResult(ErrorCode.EMPTY_ACTIVE_NODE_SET);
         }
+        int indexPos;
         String curNode;
         TcpNettyClient client;
         TcpNettyClient back1thClient = null;
         int nullClientCnt = 0;
         int unWritableCnt = 0;
         int startPos = reqSendIndex.getAndIncrement();
         for (int step = 0; step < curNodeSize; step++) {
-            curNode = curNodes.get(Math.abs(startPos++) % curNodeSize);
+            startPos = Math.abs(startPos + 1);
+            indexPos = startPos % curNodeSize;
+            if (indexPos >= curNodes.size()) {
+                if (indexExptCnt.shouldPrint()) {
+                    logger.warn("IndexOutOfBounds, startPos={}, curNodeSize={}, indexPos={}， realSize={}",
+                            startPos, curNodeSize, indexPos, curNodes.size());
+                }
+                continue;
+            }
+            curNode = curNodes.get(indexPos);
             client = usingClientMaps.get(curNode);
             if (client == null) {
                 nullClientCnt++;