From 7669ac09d685398b3e48605f3867913e73050427 Mon Sep 17 00:00:00 2001 From: waterWang Date: Thu, 27 Aug 2026 00:50:00 +0000 Subject: [PATCH] [ISSUE #10966] Fix TransactionalOpBatchService busy loop when deleteContext is drained When every entry in deleteContext has already been drained (offsets batched and sent) but is still present in the map, batchSendOpMessage() falls through to the stale lastWriteTimestamp branch and returns 0L. TransactionalOpBatchService treats 0 as "run again immediately", so waitForRunning(0) returns at once and the thread busy-loops at ~100% CPU after any transaction-message workload goes idle. Fix by returning System.currentTimeMillis() + transactionOpBatchInterval when no op message was batched in the round (sendMap == null), so the thread sleeps until the next scheduled scan. New deletes still wake it promptly via deletePrepareMessage() -> wakeup(). Signed-off-by: waterWang --- .../queue/TransactionalMessageServiceImpl.java | 11 +++++++++++ .../queue/TransactionalMessageServiceImplTest.java | 13 +++++++++++++ 2 files changed, 24 insertions(+) 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();