This is an automated email from the ASF dual-hosted git repository.
lollipopjin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new fd0c95920e [ISSUE #10827] fix(broker): spin for the lock on
same-attemptId pop orderly retry to avoid empty response (#10828)
fd0c95920e is described below
commit fd0c95920e0deac96ce2ae27442747cc5e65e930
Author: lizhimins <[email protected]>
AuthorDate: Fri Aug 7 18:29:26 2026 +0800
[ISSUE #10827] fix(broker): spin for the lock on same-attemptId pop orderly
retry to avoid empty response (#10828)
---
.../rocketmq/broker/pop/PopConsumerService.java | 28 +++-
.../pop/orderly/ConsumerOrderInfoManager.java | 12 ++
.../pop/orderly/QueueLevelConsumerManager.java | 14 ++
.../pop/PopConsumerServiceLockRetryTest.java | 158 +++++++++++++++++++++
.../pop/orderly/ConsumerOrderInfoManagerTest.java | 22 +++
5 files changed, 233 insertions(+), 1 deletion(-)
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 9ab5eb651b..f72e2ba26f 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 class PopConsumerService extends ServiceThread {
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,32 @@ public class PopConsumerService extends ServiceThread {
return getMessageFuture;
}
+ /**
+ * 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()
+ ||
!brokerController.getConsumerOrderInfoManager().isAttemptIdMatched(attemptId,
topicId, groupId)) {
+ 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<Boolean> ackAsync(
long popTime, long invisibleTime, String groupId, String topicId, int
queueId, long offset) {
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 84b0540db2..632c02053e 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 @@ public interface ConsumerOrderInfoManager {
*/
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 6f496fa13b..d1f8008a40 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 class QueueLevelConsumerManager extends
ConfigManager implements Consumer
return orderInfo.needBlock(attemptId, invisibleTime);
}
+ @Override
+ public boolean isAttemptIdMatched(String attemptId, String topic, String
group) {
+ ConcurrentHashMap<Integer/*queueId*/, OrderInfo> 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
new file mode 100644
index 0000000000..7b9f37505b
--- /dev/null
+++
b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceLockRetryTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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 ConsumerOrderInfoManager consumerOrderInfoManager;
+ 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 =
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(consumerOrderInfoManager.isAttemptIdMatched(ATTEMPT_ID,
TOPIC_ID, GROUP_ID))
+ .thenReturn(true);
+
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 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()))
+ .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());
+ }
+}
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 a5a5dfc235..557c861bd2 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 class ConsumerOrderInfoManagerTest {
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();
