[ISSUE #4349] fix negative index when index reach Integer.MAX_VALUE (#4447)

Co-authored-by: wuguoqing <wuguoqing@zhuanzhuan.com>

client/src/main/java/org/apache/rocketmq/client/common/ThreadLocalIndex.java

@@ -22,6 +22,7 @@ import java.util.Random;
 public class ThreadLocalIndex {
     private final ThreadLocal<Integer> threadLocalIndex = new ThreadLocal<Integer>();
     private final Random random = new Random();
+    private final static int POSITIVE_MASK = 0x7FFFFFFF;
 
     public int incrementAndGet() {
         Integer index = this.threadLocalIndex.get();
@@ -31,7 +32,7 @@ public class ThreadLocalIndex {
         }
 
         this.threadLocalIndex.set(++index);
-        return Math.abs(index);
+        return Math.abs(index & POSITIVE_MASK);
     }
 
     @Override

client/src/test/java/org/apache/rocketmq/client/common/ThreadLocalIndexTest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.rocketmq.client.common;
 
+import java.lang.reflect.Field;
 import org.junit.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -36,4 +37,18 @@ public class ThreadLocalIndexTest {
         assertThat(initialVal >= 0).isTrue();
     }
 
+    @Test
+    public void testIncrementAndGet3() throws Exception {
+        ThreadLocalIndex localIndex = new ThreadLocalIndex();
+        Field threadLocalIndexField = ThreadLocalIndex.class.getDeclaredField("threadLocalIndex");
+        ThreadLocal<Integer> mockThreadLocal = new ThreadLocal<Integer>();
+        mockThreadLocal.set(Integer.MAX_VALUE);
+
+        threadLocalIndexField.setAccessible(true);
+        threadLocalIndexField.set(localIndex, mockThreadLocal);
+
+        int initialVal = localIndex.incrementAndGet();
+        assertThat(initialVal >= 0).isTrue();
+    }
+
 }