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..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 @@ -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,32 @@ public CompletableFuture popAsync(String clientHost, long po 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 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 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 new file mode 100644 index 00000000000..7b9f37505b1 --- /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 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();