optimize: park merge send thread when idle to avoid 1ms polling CPU spin

core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java

@@ -82,7 +82,6 @@ public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting
     private static final String MSG_ID_PREFIX = "msgId:";
     private static final String FUTURES_PREFIX = "futures:";
     private static final String SINGLE_LOG_POSTFIX = ";";
-    private static final int MAX_MERGE_SEND_MILLS = 1;
     private static final String THREAD_PREFIX_SPLIT_CHAR = "_";
     private static final int MAX_MERGE_SEND_THREAD = 1;
     private static final long KEEP_ALIVE_TIME = Integer.MAX_VALUE;
@@ -587,7 +586,15 @@ public void run() {
             while (true) {
                 mergeLock.lock();
                 try {
-                    mergeCondition.await(MAX_MERGE_SEND_MILLS, TimeUnit.MILLISECONDS);
+                    // Park until there are pending messages, so the merge thread no longer
+                    // burns CPU with a 1ms polling cycle when idle. The check-and-wait is
+                    // atomic under mergeLock and producers offer to the basket before
+                    // signalling (see sendSyncRequest), so no wake-up can be lost.
+                    while (isBasketEmpty()) {
+                        isSending = false;
+                        mergeCondition.await();
+                    }
+                    isSending = true;
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                     LOGGER.warn("MergedSendRunnable wait interrupted", e);
@@ -639,6 +646,21 @@ public void run() {
             }
         }
 
+        /**
+         * Checks whether all baskets are empty. The merge thread parks itself
+         * when this returns true, avoiding the idle 1ms polling busy loop.
+         *
+         * @return true if every basket in basketMap is empty
+         */
+        private boolean isBasketEmpty() {
+            for (BlockingQueue<RpcMessage> basket : basketMap.values()) {
+                if (!basket.isEmpty()) {
+                    return false;
+                }
+            }
+            return true;
+        }
+
         private void printMergeMessageLog(MergedWarpMessage mergeMessage) {
             if (LOGGER.isDebugEnabled()) {
                 LOGGER.debug("merge msg size:{}", mergeMessage.msgIds.size());

core/src/test/java/org/apache/seata/core/rpc/netty/NettyRemotingClientBehaviorTest.java

@@ -38,6 +38,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
+import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -1479,6 +1480,52 @@ public void testMergedSendRunnableWithEmptyBasket() throws Exception {
         }
     }
 
+    @Test
+    public void testMergedSendRunnableIdleWaitState() throws Exception {
+        TestNettyRemotingClientWithMergeRunnable mergeClient =
+                new TestNettyRemotingClientWithMergeRunnable(clientConfig, messageExecutor);
+
+        try {
+            mergeClient.init();
+
+            // Wait for the merge send thread to start and park itself
+            Thread.sleep(300);
+
+            // The merge thread must be parked on Condition.await (WAITING) instead of
+            // spinning on a 1ms timed wait (TIMED_WAITING) when idle
+            for (Map.Entry<Thread, StackTraceElement[]> entry :
+                    Thread.getAllStackTraces().entrySet()) {
+                Thread thread = entry.getKey();
+                if (thread.getName().startsWith("rpcMergeMessageSend")) {
+                    assertFalse(
+                            Thread.State.TIMED_WAITING == thread.getState(),
+                            "merge send thread should not spin on a 1ms timed wait when idle: " + thread.getName());
+                }
+            }
+
+            // A message must wake the thread and get drained from the basket
+            GlobalBeginRequest request = new GlobalBeginRequest();
+            request.setTransactionName("test-tx-idle-wake");
+            try {
+                mergeClient.sendSyncRequest(request);
+            } catch (Exception e) {
+                // Expected: no real server at 127.0.0.1:8080, the merge thread
+                // drains the basket and fast-fails the future
+            }
+
+            Thread.sleep(300);
+            assertTrue(
+                    mergeClient.basketMap.values().stream().allMatch(BlockingQueue::isEmpty),
+                    "basket should be drained by the merge send thread after wake-up");
+        } finally {
+            try {
+                mergeClient.destroy();
+            } catch (Exception e) {
+                // Ignore
+            }
+        }
+    }
+
     /**
      * Test implementation that simulates reconnect exception
      */