From 29d7ce29488ae0494e759493d0a6699a202e770d Mon Sep 17 00:00:00 2001 From: root Date: Sun, 30 Aug 2026 01:35:31 +0800 Subject: [PATCH] [ISSUE #10989] Fix transactional message escape retry backoff (2 ^ n computed with XOR) --- .../transaction/queue/TransactionalMessageServiceImpl.java | 6 +++++- .../queue/TransactionalMessageServiceImplTest.java | 7 +++++++ 2 files changed, 12 insertions(+), 1 deletion(-) 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..17ac36f08ac 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 @@ -61,6 +61,10 @@ public class TransactionalMessageServiceImpl implements TransactionalMessageServ private static final int MAX_PROCESS_TIME_LIMIT = 60000; private static final int MAX_RETRY_TIMES_FOR_ESCAPE = 10; + static long escapeRetryBackoffMillis(int escapeFailCnt) { + return 100L * (1 << escapeFailCnt); + } + private static final int MAX_RETRY_COUNT_WHEN_HALF_NULL = 1; private static final int OP_MSG_PULL_NUMS = 32; @@ -250,7 +254,7 @@ public void check(long transactionTimeout, int transactionCheckMax, msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX)); if (escapeFailCnt < MAX_RETRY_TIMES_FOR_ESCAPE) { escapeFailCnt++; - Thread.sleep(100L * (2 ^ escapeFailCnt)); + Thread.sleep(escapeRetryBackoffMillis(escapeFailCnt)); } else { escapeFailCnt = 0; newOffset = i + 1; 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..52edf0813b5 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 @@ -84,6 +84,13 @@ public void init() { queueTransactionMsgService = new TransactionalMessageServiceImpl(bridge); } + @Test + public void testEscapeRetryBackoffMillisIsExponential() { + for (int i = 1; i <= 10; i++) { + assertThat(TransactionalMessageServiceImpl.escapeRetryBackoffMillis(i)).isEqualTo(100L * (1L << i)); + } + } + @Test public void testPrepareMessage() { MessageExtBrokerInner inner = createMessageBrokerInner();