add config for enable/disable netty isWritable check && opt for connection pool (#434)

* add config for enable/disable netty isWritable check

* fix exception info print

* opt for getConnection

* opt for connection pool

Author: GroundWu

src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableConnection.java

@@ -244,15 +244,15 @@ public void checkStatus() throws Exception {
         if (connection.getChannel() == null || !connection.getChannel().isActive()) {
             reconnect("Check connection failed for address: " + connection.getUrl());
         }
-        if (!connection.getChannel().isWritable()) {
+        if (obTable.isNettyCheckWritableEnabled() && !connection.getChannel().isWritable()) {
             LOGGER.warn("The connection might be write overflow : " + connection.getUrl());
             // Wait some interval for the case when a big package is blocking the buffer but server is ok.
             // Don't bother to call flush() here as we invoke writeAndFlush() when send request.
             Thread.sleep(obTable.getNettyBlockingWaitInterval());
             if (!connection.getChannel().isWritable()) {
                 throw new ObTableConnectionUnWritableException(
                     "Check connection failed for address: " + connection.getUrl()
-                            + ", maybe write overflow!");
+                            + ", maybe write overflow! Overflow bytes:" + connection.getChannel().bytesBeforeWritable());
             }
         }
     }

src/main/java/com/alipay/oceanbase/rpc/property/Property.java

@@ -136,6 +136,8 @@ public enum Property {
     // when a big package is blocking the buffer but the server is OK.
     NETTY_BLOCKING_WAIT_INTERVAL("bolt.netty.blocking.wait.interval", 1, "netty写缓存满后等待时间"),
 
+    NETTY_CHECK_WRITABLE_ENABLED("bolt.netty.check.writable.enabled", false, "是否启用netty写缓存可写性检查"),
+
     // [ObTable][OTHERS]
     SERVER_ENABLE_REROUTING("server.enable.rerouting", false, "开启server端的重定向回复功能"),
 

src/main/java/com/alipay/oceanbase/rpc/table/AbstractObTable.java

@@ -41,6 +41,8 @@ public abstract class AbstractObTable extends AbstractTable {
 
     protected int  nettyBlockingWaitInterval = NETTY_BLOCKING_WAIT_INTERVAL.getDefaultInt();
 
+    protected boolean nettyCheckWritableEnabled = NETTY_CHECK_WRITABLE_ENABLED.getDefaultBoolean();
+
     protected long maxConnExpiredTime        = MAX_CONN_EXPIRED_TIME.getDefaultLong();
 
     /*
@@ -162,6 +164,13 @@ public int getNettyBlockingWaitInterval() {
         return nettyBlockingWaitInterval;
     }
 
+    /*
+     * Get netty check writable enabled.
+     */
+    public boolean isNettyCheckWritableEnabled() {
+        return nettyCheckWritableEnabled;
+    }
+
     /*
      * Get connection max expired time
      */

src/main/java/com/alipay/oceanbase/rpc/table/ObTable.java

@@ -178,6 +178,8 @@ private void initProperties() {
             nettyBufferHighWatermark);
         nettyBlockingWaitInterval = parseToInt(NETTY_BLOCKING_WAIT_INTERVAL.getKey(),
             nettyBlockingWaitInterval);
+        nettyCheckWritableEnabled = parseToBoolean(NETTY_CHECK_WRITABLE_ENABLED.getKey(),
+            nettyCheckWritableEnabled);
         enableRerouting = parseToBoolean(SERVER_ENABLE_REROUTING.getKey(), enableRerouting);
         maxConnExpiredTime = parseToLong(MAX_CONN_EXPIRED_TIME.getKey(), maxConnExpiredTime);
 
@@ -842,8 +844,10 @@ private static class ObTableConnectionPool {
         private final int                                     obTableConnectionPoolSize;
         private ObTable                                       obTable;
         private volatile AtomicReference<ObTableConnection[]> connectionPool;
-        // round-robin scheduling
-        private AtomicLong                                    turn = new AtomicLong(0);
+        // counter for checkAndReconnect: records the start position of each scan
+        private AtomicLong                                    reconnectTurn = new AtomicLong(0);
+        // counter for getConnection: increments on each call to ensure random distribution
+        private AtomicLong                                    getConnectionTurn = new AtomicLong(0);
         private boolean                                       shouldStopExpand = false;
         private ScheduledFuture<?>                            expandTaskFuture = null;
         private ScheduledFuture<?>                            checkAndReconnectFuture = null;
@@ -862,11 +866,14 @@ public ObTableConnectionPool(ObTable obTable, int connectionPoolSize) {
         public void init() throws Exception {
             // only create at most 2 connections for use
             // expand other connections (if needed) in the background
+            int initConnectionNum = this.obTableConnectionPoolSize > 10 ? 10 : this.obTableConnectionPoolSize;
             connectionPool = new AtomicReference<ObTableConnection[]>();
-            ObTableConnection[] curConnectionPool = new ObTableConnection[1];
-            curConnectionPool[0] = new ObTableConnection(obTable, obTable.isOdpMode());
-            curConnectionPool[0].enableLoginWithConfigs();
-            curConnectionPool[0].init();
+            ObTableConnection[] curConnectionPool = new ObTableConnection[initConnectionNum];
+            for (int i = 0; i < initConnectionNum; i++) {
+                curConnectionPool[i] = new ObTableConnection(obTable, obTable.isOdpMode());
+                curConnectionPool[i].enableLoginWithConfigs();
+                curConnectionPool[i].init();
+            }
 
             connectionPool.set(curConnectionPool);
             // check connection pool size and expand every 3 seconds
@@ -877,16 +884,25 @@ public void init() throws Exception {
 
         /*
          * Get connection.
+         * Use counter (getConnectionTurn) that increments on each call to ensure random distribution.
+         * Guarantees to return a valid connection if one exists, and avoids multiple threads
+         * getting the same connection even when there are consecutive expired connections.
          */
         public ObTableConnection getConnection() {
             ObTableConnection[] connections = connectionPool.get();
-            long nextTurn = turn.getAndIncrement();
-            if (nextTurn == Long.MAX_VALUE) {
-                turn.set(0);
+
+            
+            // Get starting position from counter (increments on each call for randomness)
+            long startTurn = getConnectionTurn.getAndIncrement();
+            if (startTurn == Long.MAX_VALUE) {
+                getConnectionTurn.set(0);
             }
-            int round = (int) (nextTurn % connections.length);
+            int startIdx = (int) (startTurn % connections.length);
+            
+            // Traverse all connections starting from startIdx to guarantee finding a valid one
+            // This ensures we check all connections if needed, avoiding null return
             for (int i = 0; i < connections.length; i++) {
-                int idx = (round + i) % connections.length;
+                int idx = (startIdx + i) % connections.length;
                 if (!connections[idx].isExpired()) {
                     return connections[idx];
                 }
@@ -900,12 +916,13 @@ public ObTableConnection getConnection() {
          * This method will not impact the connections in use and create a reasonable number of connections.
          * */
         private void checkAndExpandPool() {
-            if (obTableConnectionPoolSize == 1 || shouldStopExpand) {
+            ObTableConnection[] curConnections = connectionPool.get();
+            if (curConnections.length == obTableConnectionPoolSize) {
                 // stop the background task if pools reach the setting
                 this.expandTaskFuture.cancel(false);
                 return;
             }
-            ObTableConnection[] curConnections = connectionPool.get();
+            
             if (curConnections.length < obTableConnectionPoolSize) {
                 int diffSize = obTableConnectionPoolSize - curConnections.length;
                 // limit expand size not too big to ensure the instant availability of connections
@@ -947,6 +964,9 @@ private void checkAndExpandPool() {
          * 2. Mark a third of the expired connections for reconnection.  
          * 3. Pause for a predefined timeout period.  
          * 4. Attempt to reconnect the marked connections.
+         * 
+         * The scan starts from the end position of the previous scan to ensure
+         * all connections are checked over time.
          **/
         private void checkAndReconnect() {
             if (obTableConnectionPoolSize == 1) {
@@ -958,24 +978,36 @@ private void checkAndReconnect() {
             // Iterate over the connection pool to identify connections that have expired
             List<Integer> expiredConnIds = new ArrayList<>();
             ObTableConnection[] connections = connectionPool.get();
-            long num = turn.get();
-            for (int i = 1; i <= connections.length; ++i) {
-                int idx = (int) ((i + num) % connections.length);
+            // Start from the end position of previous scan
+            long startPos = reconnectTurn.get();
+            for (int i = 0; i < connections.length; ++i) {
+                int idx = (int) ((i + startPos) % connections.length);
                 if (connections[idx].checkExpired()) {
                     expiredConnIds.add(idx);
                 }
             }
 
+            // Shuffle the expired connection indices to avoid consecutive connections
+            Collections.shuffle(expiredConnIds);
+
             // Mark a third of the expired connections for reconnection
             int needReconnectCount = (int) Math.ceil(expiredConnIds.size() / 3.0);
             for (int i = 0; i < needReconnectCount; i++) {
                 int idx = expiredConnIds.get(i);
                 connections[idx].setExpired(true);
             }
 
+            // Update counter to the end position of this scan for next time
+            // Use modulo to keep it within reasonable range
+            long nextStartPos = startPos + needReconnectCount;
+            if (nextStartPos >= Long.MAX_VALUE) {
+                nextStartPos = nextStartPos % connections.length;
+            }
+            reconnectTurn.set(nextStartPos);
+
             // Sleep for a predefined timeout period before attempting reconnection
             try {
-                Thread.sleep(RPC_EXECUTE_TIMEOUT.getDefaultInt());
+                Thread.sleep((long) (obTable.getObTableExecuteTimeout() * 1.5));
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
             }