diff --git a/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java b/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java index 2f05bee0040..e6a8944b263 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java +++ b/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); } diff --git a/broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImplTest.java b/broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImplTest.java index 65980756bd7..41eb2a3326f 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImplTest.java +++ b/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();