This is an automated email from the ASF dual-hosted git repository.
RongtongJin 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 a4b659b817 [ISSUE #11250] Fix Lite topic ack re-dispatching the event
while holding the pop lock (#11251)
a4b659b817 is described below
commit a4b659b817ccc5e34d15599d38b90324e9d66255
Author: Quan <[email protected]>
AuthorDate: Thu Oct 8 09:54:09 2026 +0800
[ISSUE #11250] Fix Lite topic ack re-dispatching the event while holding
the pop lock (#11251)
* [ISSUE #11250] Fix Lite topic ack re-dispatching the event while holding
the pop lock
- Decide the re-dispatch under the pop lock and dispatch after releasing it
- Add a unit test asserting the pop lock is free when the ack dispatches
* [ISSUE #11250] Re-trigger CI
---
.../broker/processor/AckMessageProcessor.java | 7 ++-
.../broker/processor/AckMessageProcessorTest.java | 53 ++++++++++++++++++++++
2 files changed, 59 insertions(+), 1 deletion(-)
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
index 036a7040e4..2653a501b6 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
@@ -538,6 +538,7 @@ public class AckMessageProcessor implements
NettyRequestProcessor {
while (!consumerLockService.tryLock(lockKey)) {
}
+ long dispatchOffset = -1L;
try {
oldOffset = consumerOffsetManager.queryOffset(group, lmqName, 0);
if (ackOffset < oldOffset) {
@@ -549,7 +550,7 @@ public class AckMessageProcessor implements
NettyRequestProcessor {
consumerOffsetManager.commitOffset("AckLiteHost", group,
lmqName, 0, nextOffset);
}
if (!consumerOrderInfoManager.checkBlock(null, lmqName, group,
0, invisibleTime)) {
-
this.brokerController.getLiteEventDispatcher().dispatch(group, lmqName, 0,
nextOffset, -1);
+ dispatchOffset = nextOffset;
}
}
if (nextOffset == -1) {
@@ -561,6 +562,10 @@ public class AckMessageProcessor implements
NettyRequestProcessor {
} finally {
consumerLockService.unlock(lockKey);
}
+ if (dispatchOffset > -1L) {
+ // Dispatch after unlock, so a pop woken by this event does not
fail on the lock still held here.
+ this.brokerController.getLiteEventDispatcher().dispatch(group,
lmqName, 0, dispatchOffset, -1);
+ }
this.brokerController.getBrokerStatsManager().incBrokerAckNums(1);
this.brokerController.getBrokerStatsManager().incGroupAckNums(group,
requestHeader.getTopic(), 1);
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/processor/AckMessageProcessorTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/processor/AckMessageProcessorTest.java
index 1add8bd8d2..c55e745ac7 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/processor/AckMessageProcessorTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/processor/AckMessageProcessorTest.java
@@ -22,10 +22,15 @@ import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.net.Broker2Client;
import org.apache.rocketmq.broker.failover.EscapeBridge;
+import org.apache.rocketmq.broker.lite.AbstractLiteLifecycleManager;
+import org.apache.rocketmq.broker.lite.LiteEventDispatcher;
import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
+import org.apache.rocketmq.broker.pop.PopConsumerLockService;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.lite.LiteUtil;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
@@ -42,6 +47,7 @@ import
org.apache.rocketmq.remoting.protocol.body.BatchAckMessageRequestBody;
import org.apache.rocketmq.remoting.protocol.header.AckMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData;
+import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.store.AppendMessageResult;
import org.apache.rocketmq.store.AppendMessageStatus;
import org.apache.rocketmq.store.DefaultMessageStore;
@@ -62,13 +68,17 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
import static
org.apache.rocketmq.broker.processor.PullMessageProcessorTest.createConsumerData;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
@@ -366,4 +376,47 @@ public class AckMessageProcessorTest {
}
}
+ @Test
+ public void testAckLite_dispatchAfterUnlock() {
+ String liteTopic = "liteTopic";
+ String lmqName = LiteUtil.toLmqName(topic, liteTopic);
+ SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+ groupConfig.setGroupName(group);
+ groupConfig.setLiteBindTopic(topic);
+
brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().put(group,
groupConfig);
+ AbstractLiteLifecycleManager liteLifecycleManager =
mock(AbstractLiteLifecycleManager.class);
+
when(liteLifecycleManager.getMaxOffsetInQueue(lmqName)).thenReturn(10L);
+
doReturn(liteLifecycleManager).when(brokerController).getLiteLifecycleManager();
+
+ // A pop woken by the ack's dispatch must be able to take the pop lock
of this lmq.
+ PopConsumerLockService lockService =
brokerController.getPopLiteMessageProcessor().getLockService();
+ String lockKey = KeyBuilder.buildPopLiteLockKey(group, lmqName);
+ AtomicBoolean lockFreeOnDispatch = new AtomicBoolean(false);
+ LiteEventDispatcher liteEventDispatcher =
mock(LiteEventDispatcher.class);
+ doAnswer(invocation -> {
+ if (lockService.tryLock(lockKey)) {
+ lockFreeOnDispatch.set(true);
+ lockService.unlock(lockKey);
+ }
+ return null;
+ }).when(liteEventDispatcher).dispatch(group, lmqName, 0, 1L, -1L);
+
doReturn(liteEventDispatcher).when(brokerController).getLiteEventDispatcher();
+
+ AckMessageRequestHeader requestHeader = new AckMessageRequestHeader();
+ requestHeader.setConsumerGroup(group);
+ requestHeader.setTopic(topic);
+ requestHeader.setLiteTopic(liteTopic);
+ requestHeader.setQueueId(0);
+ requestHeader.setOffset(0L);
+ requestHeader.setExtraInfo(ExtraInfoUtil.buildExtraInfo(0,
System.currentTimeMillis(), 60000L,
+ KeyBuilder.POP_ORDER_REVIVE_QUEUE, topic, "brokerName", 0));
+ RemotingCommand response = RemotingCommand.createResponseCommand(null);
+
+ ackMessageProcessor.ackLite(requestHeader, null, response, channel);
+
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+ verify(liteEventDispatcher).dispatch(group, lmqName, 0, 1L, -1L);
+ assertThat(lockFreeOnDispatch.get()).isTrue();
+ }
+
}
