From 3df14b93206029dde23afdd47502fdc5c5a177ff Mon Sep 17 00:00:00 2001 From: lizhimins <707364882@qq.com> Date: Fri, 7 Aug 2026 11:59:44 +0800 Subject: [PATCH 1/2] [ISSUE #10827] fix(broker): spin for the lock on same-attemptId pop orderly retry to avoid empty response An orderly retry carrying the same attemptId is an idempotent reentrant request, but the old logic fails fast with an empty response on group@topic lock contention. The retry then suspends in long polling, burns the only reentrant opportunity, times out, and the client rotates to a new attemptId, permanently losing reentrancy and blocking the queue head (up to invisibleTime, or ~3h when proxy autoRenew keeps extending nextVisibleTime). Fifo requests with a non-empty attemptId now spin-retry tryLock until the lock is acquired; other requests keep the fail-fast behavior. The lock holder always releases on pop completion, with the lock service's 2-minute expiry sweep as the worst-case backstop, so the spin cannot wait forever; same-attemptId contention is rare and the wait is normally milliseconds. Once the lock is acquired the existing re-pop path runs, fully preserving the reentrant semantics. --- .../broker/pop/PopConsumerService.java | 26 +++- .../pop/PopConsumerServiceLockRetryTest.java | 138 ++++++++++++++++++ 2 files changed, 163 insertions(+), 1 deletion(-) create mode 100644 broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java index 9ab5eb651be..1e9051b9e05 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java @@ -360,7 +360,7 @@ public CompletableFuture popAsync(String clientHost, long po new PopConsumerContext(clientHost, popTime, invisibleTime, groupId, fifo, initMode, attemptId); TopicConfig topicConfig = brokerController.getTopicConfigManager().selectTopicConfig(topicId); - if (topicConfig == null || !consumerLockService.tryLock(groupId, topicId)) { + if (topicConfig == null || !this.tryLockForPop(groupId, topicId, fifo, attemptId)) { return CompletableFuture.completedFuture(popConsumerContext); } @@ -470,6 +470,30 @@ public CompletableFuture popAsync(String clientHost, long po return getMessageFuture; } + /** + * Fifo pops carrying an attemptId are in-flight retries of the same receive attempt, + * whose batch has already been registered in OrderInfo. Instead of failing fast on + * lock contention (which leaves the retry empty and burns the reentrant attemptId), + * retry the lock briefly; other requests keep the fail-fast behavior. + */ + private boolean tryLockForPop(String groupId, String topicId, boolean fifo, String attemptId) { + if (consumerLockService.tryLock(groupId, topicId)) { + return true; + } + if (!fifo || attemptId == null || attemptId.isEmpty()) { + return false; + } + // The lock holder always releases on pop completion, and stale locks are + // removed by PopConsumerLockService.removeTimeout(), so keep retrying until + // the lock is acquired to make sure the in-flight retry is not left empty. + // Same-attemptId contention is rare and the wait is normally milliseconds, + // so a plain spin is fine here. + while (!consumerLockService.tryLock(groupId, topicId)) { + Thread.yield(); + } + return true; + } + // Notify polling request when receive orderly ack public CompletableFuture ackAsync( long popTime, long invisibleTime, String groupId, String topicId, int queueId, long offset) { diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java new file mode 100644 index 00000000000..46dd0b9587e --- /dev/null +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.rocketmq.broker.pop; + +import java.io.File; +import java.io.IOException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.commons.io.FileUtils; +import org.apache.commons.lang3.reflect.FieldUtils; +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; +import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager; +import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager; +import org.apache.rocketmq.broker.topic.TopicConfigManager; +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.TopicConfig; +import org.apache.rocketmq.common.constant.ConsumeInitMode; +import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; +import org.apache.rocketmq.store.GetMessageResult; +import org.apache.rocketmq.store.GetMessageStatus; +import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +public class PopConsumerServiceLockRetryTest { + + private static final String GROUP_ID = "groupId"; + private static final String TOPIC_ID = "topicId"; + private static final String ATTEMPT_ID = "attempt-id-lock-retry"; + private static final long INVISIBLE_TIME = 300_000L; + + private final String filePath = PopConsumerRocksdbStoreTest.getRandomStorePath(); + + private BrokerController brokerController; + private PopConsumerLockService consumerLockService; + private SubscriptionGroupManager subscriptionGroupManager; + private ConsumerOffsetManager consumerOffsetManager; + private MessageStore messageStore; + private PopConsumerService consumerService; + + @Before + public void init() throws IOException, IllegalAccessException { + BrokerConfig brokerConfig = new BrokerConfig(); + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + messageStoreConfig.setStorePathRootDir(filePath); + + TopicConfigManager topicConfigManager = Mockito.mock(TopicConfigManager.class); + subscriptionGroupManager = Mockito.mock(SubscriptionGroupManager.class); + consumerOffsetManager = Mockito.mock(ConsumerOffsetManager.class); + ConsumerOrderInfoManager consumerOrderInfoManager = Mockito.mock(ConsumerOrderInfoManager.class); + consumerLockService = Mockito.mock(PopConsumerLockService.class); + messageStore = Mockito.mock(MessageStore.class); + + brokerController = Mockito.mock(BrokerController.class); + Mockito.when(brokerController.getBrokerConfig()).thenReturn(brokerConfig); + Mockito.when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig); + Mockito.when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager); + Mockito.when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager); + Mockito.when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager); + Mockito.when(brokerController.getConsumerOrderInfoManager()).thenReturn(consumerOrderInfoManager); + Mockito.when(brokerController.getMessageStore()).thenReturn(messageStore); + Mockito.when(topicConfigManager.selectTopicConfig(Mockito.anyString())) + .thenReturn(new TopicConfig(TOPIC_ID)); + + consumerService = new PopConsumerService(brokerController); + // the lock service is built inside the constructor, replace it for verification + FieldUtils.writeField(consumerService, "consumerLockService", consumerLockService, true); + } + + @After + public void shutdown() throws IOException { + FileUtils.deleteDirectory(new File(filePath)); + } + + private void stubEmptyStore() { + GetMessageResult result = new GetMessageResult(); + result.setStatus(GetMessageStatus.NO_MESSAGE_IN_QUEUE); + result.setNextBeginOffset(0L); + Mockito.when(messageStore.getMessageAsync(Mockito.anyString(), Mockito.anyString(), + Mockito.anyInt(), Mockito.anyLong(), Mockito.anyInt(), Mockito.any())) + .thenReturn(CompletableFuture.completedFuture(result)); + Mockito.when(consumerOffsetManager.queryOffset(Mockito.anyString(), Mockito.anyString(), + Mockito.anyInt())).thenReturn(0L); + } + + @Test + public void popAsyncOrderlyLockRetryPersistsTest() { + // the retry keeps spinning until the lock holder releases, no early give-up + AtomicInteger attempts = new AtomicInteger(); + Mockito.when(consumerLockService.tryLock(Mockito.anyString(), Mockito.anyString())) + .thenAnswer(invocation -> attempts.incrementAndGet() > 50); + Mockito.when(subscriptionGroupManager.findSubscriptionGroupConfig(Mockito.anyString())) + .thenReturn(new SubscriptionGroupConfig()); + stubEmptyStore(); + + PopConsumerContext result = consumerService.popAsync("127.0.0.1", System.currentTimeMillis(), + INVISIBLE_TIME, GROUP_ID, TOPIC_ID, 0, 32, true, ATTEMPT_ID, ConsumeInitMode.MIN, null).join(); + + assertNotNull(result); + Mockito.verify(consumerLockService, Mockito.times(51)).tryLock(Mockito.anyString(), Mockito.anyString()); + Mockito.verify(subscriptionGroupManager).findSubscriptionGroupConfig(GROUP_ID); + } + + @Test + public void popAsyncNonFifoFailFastTest() { + Mockito.when(consumerLockService.tryLock(Mockito.anyString(), Mockito.anyString())) + .thenReturn(false); + + PopConsumerContext result = consumerService.popAsync("127.0.0.1", System.currentTimeMillis(), + INVISIBLE_TIME, GROUP_ID, TOPIC_ID, 0, 32, false, null, ConsumeInitMode.MIN, null).join(); + + assertNotNull(result); + assertEquals(0, result.getMessageCount()); + // non-fifo requests must keep the fail-fast behavior + Mockito.verify(consumerLockService, Mockito.times(1)).tryLock(Mockito.anyString(), Mockito.anyString()); + } +} From 43461f940c3478b0ac9210ff712fe786d5bf5931 Mon Sep 17 00:00:00 2001 From: lizhimins <707364882@qq.com> Date: Fri, 7 Aug 2026 15:50:57 +0800 Subject: [PATCH 2/2] [ISSUE #10827] fix(broker): only spin for the lock when the attemptId is registered in OrderInfo Fifo pops with different attemptIds would be blocked by checkBlock even after acquiring the lock, so spinning on contention only wastes request threads while the lock may be held for seconds on slow paths. Restrict the forced lock acquisition to requests whose attemptId is already registered in OrderInfo, i.e. genuine in-flight retries of a previous delivery of the same receive attempt. Add ConsumerOrderInfoManager.isAttemptIdMatched to look up the attemptId across the queues of the topic@group. --- .../broker/pop/PopConsumerService.java | 12 +++++----- .../pop/orderly/ConsumerOrderInfoManager.java | 12 ++++++++++ .../orderly/QueueLevelConsumerManager.java | 14 ++++++++++++ .../pop/PopConsumerServiceLockRetryTest.java | 22 ++++++++++++++++++- .../orderly/ConsumerOrderInfoManagerTest.java | 22 +++++++++++++++++++ 5 files changed, 76 insertions(+), 6 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java index 1e9051b9e05..f72e2ba26f2 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java @@ -471,16 +471,18 @@ public CompletableFuture popAsync(String clientHost, long po } /** - * Fifo pops carrying an attemptId are in-flight retries of the same receive attempt, - * whose batch has already been registered in OrderInfo. Instead of failing fast on - * lock contention (which leaves the retry empty and burns the reentrant attemptId), - * retry the lock briefly; other requests keep the fail-fast behavior. + * Fifo pops carrying an attemptId already registered in OrderInfo are in-flight retries + * of the same receive attempt. Instead of failing fast on lock contention (which leaves + * the retry empty and burns the reentrant attemptId), keep retrying the lock; pops with + * a different attemptId would be blocked by checkBlock anyway, so they keep the + * fail-fast behavior. */ private boolean tryLockForPop(String groupId, String topicId, boolean fifo, String attemptId) { if (consumerLockService.tryLock(groupId, topicId)) { return true; } - if (!fifo || attemptId == null || attemptId.isEmpty()) { + if (!fifo || attemptId == null || attemptId.isEmpty() + || !brokerController.getConsumerOrderInfoManager().isAttemptIdMatched(attemptId, topicId, groupId)) { return false; } // The lock holder always releases on pop completion, and stale locks are diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManager.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManager.java index 84b0540db24..632c02053e0 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManager.java @@ -68,6 +68,18 @@ void update(String attemptId, boolean isRetry, String topic, String group, int q */ boolean checkBlock(String attemptId, String topic, String group, int queueId, long invisibleTime); + /** + * Check whether the given attemptId has already been registered in the order info of + * the topic and group, i.e. the request is an in-flight retry of a previous delivery + * of the same receive attempt + * + * @param attemptId Attempt ID + * @param topic Topic name + * @param group Consumer group name + * @return true indicates the attemptId is registered in some queue's order info + */ + boolean isAttemptIdMatched(String attemptId, String topic, String group); + /** * Remove the specified topic and group * Usually called during topic deletion diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java index 6f496fa13b3..d1f8008a406 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java @@ -171,6 +171,20 @@ public boolean checkBlock(String attemptId, String topic, String group, int queu return orderInfo.needBlock(attemptId, invisibleTime); } + @Override + public boolean isAttemptIdMatched(String attemptId, String topic, String group) { + ConcurrentHashMap qs = table.get(buildKey(topic, group)); + if (qs == null || attemptId == null) { + return false; + } + for (OrderInfo orderInfo : qs.values()) { + if (orderInfo != null && attemptId.equals(orderInfo.getAttemptId())) { + return true; + } + } + return false; + } + @Override public void clearBlock(String topic, String group, int queueId) { table.computeIfPresent(buildKey(topic, group), (key, val) -> { diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java index 46dd0b9587e..7b9f37505b1 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java @@ -56,6 +56,7 @@ public class PopConsumerServiceLockRetryTest { private PopConsumerLockService consumerLockService; private SubscriptionGroupManager subscriptionGroupManager; private ConsumerOffsetManager consumerOffsetManager; + private ConsumerOrderInfoManager consumerOrderInfoManager; private MessageStore messageStore; private PopConsumerService consumerService; @@ -68,7 +69,7 @@ public void init() throws IOException, IllegalAccessException { TopicConfigManager topicConfigManager = Mockito.mock(TopicConfigManager.class); subscriptionGroupManager = Mockito.mock(SubscriptionGroupManager.class); consumerOffsetManager = Mockito.mock(ConsumerOffsetManager.class); - ConsumerOrderInfoManager consumerOrderInfoManager = Mockito.mock(ConsumerOrderInfoManager.class); + consumerOrderInfoManager = Mockito.mock(ConsumerOrderInfoManager.class); consumerLockService = Mockito.mock(PopConsumerLockService.class); messageStore = Mockito.mock(MessageStore.class); @@ -110,6 +111,8 @@ public void popAsyncOrderlyLockRetryPersistsTest() { AtomicInteger attempts = new AtomicInteger(); Mockito.when(consumerLockService.tryLock(Mockito.anyString(), Mockito.anyString())) .thenAnswer(invocation -> attempts.incrementAndGet() > 50); + Mockito.when(consumerOrderInfoManager.isAttemptIdMatched(ATTEMPT_ID, TOPIC_ID, GROUP_ID)) + .thenReturn(true); Mockito.when(subscriptionGroupManager.findSubscriptionGroupConfig(Mockito.anyString())) .thenReturn(new SubscriptionGroupConfig()); stubEmptyStore(); @@ -122,6 +125,23 @@ public void popAsyncOrderlyLockRetryPersistsTest() { Mockito.verify(subscriptionGroupManager).findSubscriptionGroupConfig(GROUP_ID); } + @Test + public void popAsyncUnregisteredAttemptIdFailFastTest() { + // a fifo pop whose attemptId is not registered in OrderInfo is not an in-flight + // retry of a previous delivery, so it must keep the fail-fast behavior + Mockito.when(consumerLockService.tryLock(Mockito.anyString(), Mockito.anyString())) + .thenReturn(false); + Mockito.when(consumerOrderInfoManager.isAttemptIdMatched(ATTEMPT_ID, TOPIC_ID, GROUP_ID)) + .thenReturn(false); + + PopConsumerContext result = consumerService.popAsync("127.0.0.1", System.currentTimeMillis(), + INVISIBLE_TIME, GROUP_ID, TOPIC_ID, 0, 32, true, ATTEMPT_ID, ConsumeInitMode.MIN, null).join(); + + assertNotNull(result); + assertEquals(0, result.getMessageCount()); + Mockito.verify(consumerLockService, Mockito.times(1)).tryLock(Mockito.anyString(), Mockito.anyString()); + } + @Test public void popAsyncNonFifoFailFastTest() { Mockito.when(consumerLockService.tryLock(Mockito.anyString(), Mockito.anyString())) diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManagerTest.java index a5a5dfc2357..557c861bd2b 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManagerTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManagerTest.java @@ -535,6 +535,28 @@ public void testReentrant() { assertFalse(consumerOrderInfoManager.checkBlock(attemptId, TOPIC, GROUP, QUEUE_ID_0, 3000)); } + @Test + public void isAttemptIdMatchTest() { + StringBuilder orderInfoBuilder = new StringBuilder(); + String attemptId = UUID.randomUUID().toString(); + consumerOrderInfoManager.update( + attemptId, + false, + TOPIC, + GROUP, + QUEUE_ID_0, + popTime, + 3000, + Lists.newArrayList(1L, 2L, 3L), + orderInfoBuilder + ); + + assertTrue(consumerOrderInfoManager.isAttemptIdMatched(attemptId, TOPIC, GROUP)); + assertFalse(consumerOrderInfoManager.isAttemptIdMatched(UUID.randomUUID().toString(), TOPIC, GROUP)); + assertFalse(consumerOrderInfoManager.isAttemptIdMatched(attemptId, "unknownTopic", GROUP)); + assertFalse(consumerOrderInfoManager.isAttemptIdMatched(null, TOPIC, GROUP)); + } + @Test public void testGetMaxLockFreeTimestamp() { QueueLevelConsumerManager.OrderInfo orderInfo = new QueueLevelConsumerManager.OrderInfo();