[ISSUE #10966] Fix TransactionalOpBatchService busy loop when deleteContext is drained

broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java

@@ -745,6 +745,17 @@ public long batchSendOpMessage() {
             if (!overSize && wakeupTimestamp > startTime) {
                 return wakeupTimestamp;
             }
+
+            // Nothing was batched in this round (e.g. every context in deleteContext
+            // is already drained but still present, so firstTimestamp falls back to a
+            // stale lastWriteTimestamp and the branch above does not fire). Returning 0
+            // would make TransactionalOpBatchService treat it as "run again immediately",
+            // busy-looping one CPU core at ~100%. Schedule the next scan after the batch
+            // interval instead; new deletes still wake the thread promptly via
+            // deletePrepareMessage() -> wakeup().
+            if (sendMap == null) {
+                return System.currentTimeMillis() + interval;
+            }
         } catch (Throwable t) {
             log.error("batchSendOp error.", t);
         }

broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImplTest.java

@@ -173,6 +173,19 @@ public void testDeletePrepareMessage_maxSize() throws InterruptedException {
         queueTransactionMsgService.close();
     }
 
+    @Test
+    public void testBatchSendOpMessage_noPendingDataReturnsFutureWakeup() {
+        // Reproduce #10966: a drained context (all offsets already batched and sent)
+        // stays in deleteContext forever with a stale lastWriteTimestamp. Previously
+        // batchSendOpMessage() returned 0 in this state, which made
+        // TransactionalOpBatchService busy-loop at ~100% CPU.
+        ((TransactionalMessageServiceImpl) queueTransactionMsgService).getDeleteContext()
+            .put(0, new MessageQueueOpContext(0L, 1));
+        long wakeup = ((TransactionalMessageServiceImpl) queueTransactionMsgService).batchSendOpMessage();
+        // Should schedule the next scan after the batch interval, not 0.
+        assertThat(wakeup).isGreaterThan(System.currentTimeMillis());
+    }
+
     @Test
     public void testOpen() {
         boolean isOpen = queueTransactionMsgService.open();