From 5fb35b4018b9b55e4b8ab85248467df03d20e7c4 Mon Sep 17 00:00:00 2001 From: redlsz Date: Mon, 24 Aug 2026 14:15:38 +0800 Subject: [PATCH 1/4] fix #ISSUE 10949 --- .../rocketmq/broker/pop/PopConsumerCache.java | 48 ++++ .../broker/pop/PopConsumerContext.java | 40 +++ .../broker/pop/PopConsumerService.java | 57 ++++- .../broker/pop/PopConsumerCacheTest.java | 63 +++++ .../PopConsumerServiceCommitOffsetTest.java | 242 ++++++++++++++++++ 5 files changed, 441 insertions(+), 9 deletions(-) create mode 100644 broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceCommitOffsetTest.java diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java index c74c5793a5c..e6740ad29d5 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java @@ -84,6 +84,13 @@ public long getMinOffsetInCache(String groupId, String topicId, int queueId) { return consumerRecords != null ? consumerRecords.getMinOffsetInBuffer() : OFFSET_NOT_EXIST; } + public void setPendingCommitOffset(String groupId, String topicId, int queueId, long offset) { + ConsumerRecords consumerRecords = consumerRecordTable.get(this.getKey(groupId, topicId, queueId)); + if (consumerRecords != null) { + consumerRecords.setPendingCommitOffset(offset); + } + } + public long getPopInFlightMessageCount(String groupId, String topicId, int queueId) { ConsumerRecords consumerRecords = consumerRecordTable.get(this.getKey(groupId, topicId, queueId)); return consumerRecords != null ? consumerRecords.getInFlightRecordCount() : 0L; @@ -133,6 +140,7 @@ public int cleanupRecords(Consumer consumer) { consumerRecordStore.writeRecords(writeConsumerRecords); } records.clearStagedRecords(); + this.commitPendingOffset(records); log.info("PopConsumerOffline, so clean expire records, groupId={}, topic={}, queueId={}, records={}", records.getGroupId(), records.getTopicId(), records.getQueueId(), records.getInFlightRecordCount()); iterator.remove(); @@ -159,6 +167,8 @@ public int cleanupRecords(Consumer consumer) { if (offset > OFFSET_NOT_EXIST) { this.commitOffset("PopConsumerCache", records.getGroupId(), records.getTopicId(), records.getQueueId(), offset); + } else { + this.commitPendingOffset(records); } remain += records.getInFlightRecordCount(); @@ -166,6 +176,33 @@ public int cleanupRecords(Consumer consumer) { return remain; } + /** + * Commit the offset left over by pop once no record of the queue remains in the cache. + */ + private void commitPendingOffset(ConsumerRecords records) { + // Read the pending offset before checking the in-flight records, never the other way + // around. A concurrent pop writes its records into the cache before it updates the pending + // offset, so this order guarantees that either the check below sees those records, or the + // pending offset read here is the older one. + long pendingCommitOffset = records.getPendingCommitOffset(); + if (pendingCommitOffset == OFFSET_NOT_EXIST || records.getInFlightRecordCount() != 0) { + return; + } + + String groupId = records.getGroupId(); + String topicId = records.getTopicId(); + int queueId = records.getQueueId(); + + ConsumerOffsetManager consumerOffsetManager = brokerController.getConsumerOffsetManager(); + if (consumerOffsetManager.hasOffsetReset(topicId, groupId, queueId)) { + return; + } + + if (pendingCommitOffset > consumerOffsetManager.queryOffset(groupId, topicId, queueId)) { + this.commitOffset("PopConsumerCache", groupId, topicId, queueId, pendingCommitOffset); + } + } + public void commitOffset(String clientHost, String groupId, String topicId, int queueId, long offset) { if (!consumerLockService.tryLock(groupId, topicId)) { return; @@ -214,6 +251,9 @@ protected static class ConsumerRecords { private final ConcurrentSkipListMap removeTreeMap; private final ConcurrentSkipListMap recordTreeMap; + // The consumer offset to commit once no record of this queue is in cache + private volatile long pendingCommitOffset = OFFSET_NOT_EXIST; + public ConsumerRecords(BrokerConfig brokerConfig, String groupId, String topicId, int queueId) { this.groupId = groupId; this.topicId = topicId; @@ -244,6 +284,14 @@ public int getInFlightRecordCount() { return removeTreeMap.size() + recordTreeMap.size(); } + public void setPendingCommitOffset(long pendingCommitOffset) { + this.pendingCommitOffset = pendingCommitOffset; + } + + public long getPendingCommitOffset() { + return pendingCommitOffset; + } + public void stageExpiredRecords(long currentTime) { Iterator> iterator = recordTreeMap.entrySet().iterator(); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerContext.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerContext.java index 0ad8bacab1c..bc0461a521e 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerContext.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerContext.java @@ -51,6 +51,8 @@ public class PopConsumerContext { private List popConsumerRecordList; + private List pendingCommitList; + public PopConsumerContext(String clientHost, long popTime, long invisibleTime, String groupId, boolean fifo, int initMode, String attemptId) { @@ -164,6 +166,44 @@ public List getPopConsumerRecordList() { return popConsumerRecordList; } + public void addPendingCommit(String topicId, int queueId, long commitOffset) { + if (this.pendingCommitList == null) { + this.pendingCommitList = new ArrayList<>(); + } + this.pendingCommitList.add(new PendingCommit(topicId, queueId, commitOffset)); + } + + public List getPendingCommitList() { + return pendingCommitList; + } + + public static class PendingCommit { + + private final String topicId; + + private final int queueId; + + private final long commitOffset; + + public PendingCommit(String topicId, int queueId, long commitOffset) { + this.topicId = topicId; + this.queueId = queueId; + this.commitOffset = commitOffset; + } + + public String getTopicId() { + return topicId; + } + + public int getQueueId() { + return queueId; + } + + public long getCommitOffset() { + return commitOffset; + } + } + @Override public String toString() { return "PopConsumerContext{" + 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 f72e2ba26f2..0397fde05f2 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 @@ -39,6 +39,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.tuple.Triple; import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.KeyBuilder; import org.apache.rocketmq.common.MixAll; @@ -185,23 +186,18 @@ public PopConsumerContext handleGetMessageResult(PopConsumerContext context, Get } } - long commitOffset = offset; if (context.isFifo()) { + long commitOffset = offset; if (!GetMessageStatus.FOUND.equals(result.getStatus())) { commitOffset = result.getNextBeginOffset(); } + this.brokerController.getConsumerOffsetManager().commitOffset( + context.getClientHost(), context.getGroupId(), topicId, queueId, commitOffset); } else { this.brokerController.getConsumerOffsetManager().commitPullOffset( context.getClientHost(), context.getGroupId(), topicId, queueId, result.getNextBeginOffset()); - if (brokerConfig.isEnablePopBufferMerge() && popConsumerCache != null) { - long minOffset = popConsumerCache.getMinOffsetInCache(context.getGroupId(), topicId, queueId); - if (minOffset != OFFSET_NOT_EXIST) { - commitOffset = minOffset; - } - } + context.addPendingCommit(topicId, queueId, result.getNextBeginOffset()); } - this.brokerController.getConsumerOffsetManager().commitOffset( - context.getClientHost(), context.getGroupId(), topicId, queueId, commitOffset); return context; } @@ -447,6 +443,7 @@ public CompletableFuture popAsync(String clientHost, long po } } } + this.commitPendingOffset(result); return CompletableFuture.completedFuture(result); }).whenComplete((result, throwable) -> { try { @@ -470,6 +467,48 @@ public CompletableFuture popAsync(String clientHost, long po return getMessageFuture; } + /** + * Commit the consumer offset up to nextBeginOffset of this pop, instead of the batch start + * offset which always lags one round behind. Acknowledgements only delete records, so the offset would + * otherwise never catch up once a consumer acknowledged everything and went offline. + *

+ * Records written to the kv store are durable and unacknowledged ones are redelivered by revive, so + * the offset may pass them. Records held in the cache are not persisted yet, so they cap the + * offset at minOffsetInCache here, and nextBeginOffset is kept on the cache entry for + * PopConsumerCache#cleanupRecords to commit once they are acknowledged or persisted. + */ + protected void commitPendingOffset(PopConsumerContext context) { + List pendingCommitList = context.getPendingCommitList(); + if (pendingCommitList == null) { + return; + } + + String groupId = context.getGroupId(); + + for (PopConsumerContext.PendingCommit pendingCommit : pendingCommitList) { + String topicId = pendingCommit.getTopicId(); + int queueId = pendingCommit.getQueueId(); + + ConsumerOffsetManager consumerOffsetManager = this.brokerController.getConsumerOffsetManager(); + if (consumerOffsetManager.hasOffsetReset(topicId, groupId, queueId)) { + continue; + } + + long commitOffset = pendingCommit.getCommitOffset(); + if (popConsumerCache != null) { + long minOffset = popConsumerCache.getMinOffsetInCache(groupId, topicId, queueId); + if (minOffset != OFFSET_NOT_EXIST) { + popConsumerCache.setPendingCommitOffset(groupId, topicId, queueId, commitOffset); + commitOffset = Math.min(commitOffset, minOffset); + } + } + + if (commitOffset > consumerOffsetManager.queryOffset(groupId, topicId, queueId)) { + consumerOffsetManager.commitOffset(context.getClientHost(), groupId, topicId, queueId, commitOffset); + } + } + } + /** * 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 diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java index 28045ca26e7..a9df88a0ab8 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java @@ -30,6 +30,9 @@ import org.mockito.Mockito; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; public class PopConsumerCacheTest { @@ -143,4 +146,64 @@ record = new PopConsumerRecord(2L, groupId, topicId, queueId, consumerCache.cleanupRecords(consumerRecordList::add); Assert.assertEquals(0, consumerRecordList.size()); } + + @Test + public void commitPendingOffsetTest() { + BrokerController brokerController = Mockito.mock(BrokerController.class); + PopConsumerKVStore consumerKVStore = Mockito.mock(PopConsumerRocksdbStore.class); + PopConsumerLockService consumerLockService = Mockito.mock(PopConsumerLockService.class); + ConsumerOffsetManager consumerOffsetManager = Mockito.mock(ConsumerOffsetManager.class); + Mockito.when(brokerController.getBrokerConfig()).thenReturn(new BrokerConfig()); + Mockito.when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager); + Mockito.when(consumerLockService.tryLock(groupId, topicId)).thenReturn(true); + + PopConsumerCache consumerCache = + new PopConsumerCache(brokerController, consumerKVStore, consumerLockService, null); + + // the record is still in cache, so it bounds the commit + PopConsumerRecord record = new PopConsumerRecord(System.currentTimeMillis(), + groupId, topicId, queueId, 0, 20000, 100, attemptId); + consumerCache.writeRecords(Collections.singletonList(record)); + consumerCache.setPendingCommitOffset(groupId, topicId, queueId, 110L); + consumerCache.cleanupRecords(consumerRecord -> { + }); + Mockito.verify(consumerOffsetManager).commitOffset( + anyString(), eq(groupId), eq(topicId), eq(queueId), eq(100L)); + + // nothing remains in cache after the acknowledgement, so the pending offset is committed + Assert.assertTrue(consumerCache.deleteRecords(Collections.singletonList(record)).isEmpty()); + consumerCache.cleanupRecords(consumerRecord -> { + }); + Mockito.verify(consumerOffsetManager).commitOffset( + anyString(), eq(groupId), eq(topicId), eq(queueId), eq(110L)); + + // the offset store has caught up, so the pending offset is not committed again + Mockito.when(consumerOffsetManager.queryOffset(groupId, topicId, queueId)).thenReturn(110L); + consumerCache.cleanupRecords(consumerRecord -> { + }); + Mockito.verify(consumerOffsetManager, Mockito.times(1)).commitOffset( + anyString(), eq(groupId), eq(topicId), eq(queueId), eq(110L)); + + // reset offset wins over the pending offset + Mockito.when(consumerOffsetManager.hasOffsetReset(topicId, groupId, queueId)).thenReturn(true); + consumerCache.setPendingCommitOffset(groupId, topicId, queueId, 120L); + consumerCache.cleanupRecords(consumerRecord -> { + }); + Mockito.verify(consumerOffsetManager, Mockito.never()).commitOffset( + anyString(), anyString(), anyString(), anyInt(), eq(120L)); + + // records of an offline consumer are persisted before the entry is dropped + Mockito.when(consumerOffsetManager.hasOffsetReset(topicId, groupId, queueId)).thenReturn(false); + Mockito.when(consumerLockService.isLockTimeout(any(), any())).thenReturn(true); + consumerCache.writeRecords(Collections.singletonList(new PopConsumerRecord( + System.currentTimeMillis(), groupId, topicId, queueId, 0, 20000, 200, attemptId))); + consumerCache.setPendingCommitOffset(groupId, topicId, queueId, 210L); + consumerCache.cleanupRecords(consumerRecord -> { + }); + Mockito.verify(consumerKVStore).writeRecords(Mockito.argThat(records -> + records.size() == 1 && records.get(0).getOffset() == 200L)); + Mockito.verify(consumerOffsetManager).commitOffset( + anyString(), eq(groupId), eq(topicId), eq(queueId), eq(210L)); + Assert.assertEquals(0, consumerCache.getCacheKeySize()); + } } \ No newline at end of file diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceCommitOffsetTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceCommitOffsetTest.java new file mode 100644 index 00000000000..5b9628adbbf --- /dev/null +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceCommitOffsetTest.java @@ -0,0 +1,242 @@ +/* + * 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.Collections; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +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.processor.PopMessageProcessor; +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.constant.ConsumeInitMode; +import org.apache.rocketmq.store.GetMessageResult; +import org.apache.rocketmq.store.GetMessageStatus; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.junit.After; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; + +public class PopConsumerServiceCommitOffsetTest { + + private static final long INVISIBLE_TIME = TimeUnit.SECONDS.toMillis(20); + + private final String clientHost = "127.0.0.1:8888"; + private final String groupId = "groupId"; + private final String topicId = "topicId"; + private final int queueId = 2; + private final String attemptId = UUID.randomUUID().toString().toUpperCase(); + private final String filePath = PopConsumerRocksdbStoreTest.getRandomStorePath(); + + private BrokerController brokerController; + private ConsumerOffsetManager consumerOffsetManager; + private PopConsumerService consumerService; + + private void init(boolean enablePopBufferMerge) { + BrokerConfig brokerConfig = new BrokerConfig(); + brokerConfig.setEnablePopBufferMerge(enablePopBufferMerge); + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + messageStoreConfig.setStorePathRootDir(filePath); + + consumerOffsetManager = Mockito.mock(ConsumerOffsetManager.class); + PopMessageProcessor popMessageProcessor = Mockito.mock(PopMessageProcessor.class); + + brokerController = Mockito.mock(BrokerController.class); + Mockito.when(brokerController.getBrokerConfig()).thenReturn(brokerConfig); + Mockito.when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig); + Mockito.when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager); + Mockito.when(brokerController.getTopicConfigManager()).thenReturn(Mockito.mock(TopicConfigManager.class)); + Mockito.when(brokerController.getSubscriptionGroupManager()) + .thenReturn(Mockito.mock(SubscriptionGroupManager.class)); + Mockito.when(brokerController.getPopMessageProcessor()).thenReturn(popMessageProcessor); + Mockito.when(brokerController.getConsumerOrderInfoManager()) + .thenReturn(Mockito.mock(ConsumerOrderInfoManager.class)); + + consumerService = new PopConsumerService(brokerController); + } + + @After + public void shutdown() throws IOException { + FileUtils.deleteDirectory(new File(filePath)); + } + + private PopConsumerContext popContext(boolean fifo) { + return new PopConsumerContext(clientHost, System.currentTimeMillis(), + INVISIBLE_TIME, groupId, fifo, ConsumeInitMode.MIN, attemptId); + } + + private GetMessageResult foundResult(long nextBeginOffset) { + GetMessageResult result = new GetMessageResult(); + result.setStatus(GetMessageStatus.FOUND); + result.setNextBeginOffset(nextBeginOffset); + result.getMessageQueueOffset().add(nextBeginOffset - 1); + return result; + } + + private PopConsumerCache getConsumerCache() throws IllegalAccessException { + return (PopConsumerCache) FieldUtils.readField(consumerService, "popConsumerCache", true); + } + + private void verifyCommitted(long offset) { + Mockito.verify(consumerOffsetManager).commitOffset( + anyString(), eq(groupId), eq(topicId), eq(queueId), eq(offset)); + } + + private void verifyNeverCommitted() { + Mockito.verify(consumerOffsetManager, Mockito.never()).commitOffset( + anyString(), anyString(), anyString(), anyInt(), anyLong()); + } + + @Test + public void commitToNextBeginOffsetWithoutCache() { + init(false); + consumerService.getPopConsumerStore().start(); + + PopConsumerContext context = popContext(false); + consumerService.handleGetMessageResult(context, foundResult(110L), + topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100L); + + // staged only, not committed before the records are written + verifyNeverCommitted(); + Assert.assertEquals(1, context.getPendingCommitList().size()); + Assert.assertEquals(110L, context.getPendingCommitList().get(0).getCommitOffset()); + + consumerService.getPopConsumerStore().writeRecords(context.getPopConsumerRecordList()); + consumerService.commitPendingOffset(context); + verifyCommitted(110L); + + consumerService.shutdown(); + } + + @Test + public void commitPullOffsetAlways() { + init(false); + PopConsumerContext context = popContext(false); + consumerService.handleGetMessageResult(context, foundResult(110L), + topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100L); + Mockito.verify(consumerOffsetManager).commitPullOffset( + anyString(), eq(groupId), eq(topicId), eq(queueId), eq(110L)); + } + + @Test + public void commitWhenNoMatchedMessage() { + init(false); + GetMessageResult result = new GetMessageResult(); + result.setStatus(GetMessageStatus.NO_MATCHED_MESSAGE); + result.setNextBeginOffset(110L); + + PopConsumerContext context = popContext(false); + consumerService.handleGetMessageResult(context, result, + topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100L); + Assert.assertFalse(context.isFound()); + + consumerService.commitPendingOffset(context); + verifyCommitted(110L); + } + + @Test + public void commitBoundedByMinOffsetInCache() throws IllegalAccessException { + init(true); + PopConsumerCache consumerCache = getConsumerCache(); + Assert.assertNotNull(consumerCache); + + PopConsumerContext context = popContext(false); + consumerService.handleGetMessageResult(context, foundResult(110L), + topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100L); + consumerCache.writeRecords(Collections.singletonList(new PopConsumerRecord( + System.currentTimeMillis(), groupId, topicId, queueId, 0, INVISIBLE_TIME, 105L, attemptId))); + + consumerService.commitPendingOffset(context); + + // bounded by the in-flight record at 105 rather than jumping to 110 + verifyCommitted(105L); + // the remaining part converges in PopConsumerCache#cleanupRecords once the record is acked + Assert.assertEquals(105L, consumerCache.getMinOffsetInCache(groupId, topicId, queueId)); + } + + @Test + public void commitBoundedWhenCacheFull() throws IllegalAccessException { + init(true); + PopConsumerCache consumerCache = getConsumerCache(); + consumerCache.writeRecords(Collections.singletonList(new PopConsumerRecord( + System.currentTimeMillis(), groupId, topicId, queueId, 0, INVISIBLE_TIME, 100L, attemptId))); + + // the later batch bypassed the cache, but the cached record at 100 is still in flight + PopConsumerContext context = popContext(false); + consumerService.handleGetMessageResult(context, foundResult(120L), + topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 110L); + consumerService.commitPendingOffset(context); + + verifyCommitted(100L); + } + + @Test + public void skipCommitWhenOffsetReset() { + init(false); + Mockito.when(consumerOffsetManager.hasOffsetReset(topicId, groupId, queueId)).thenReturn(true); + + PopConsumerContext context = popContext(false); + consumerService.handleGetMessageResult(context, foundResult(110L), + topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100L); + consumerService.commitPendingOffset(context); + + verifyNeverCommitted(); + } + + @Test + public void commitIsIdempotent() { + init(false); + PopConsumerContext context = popContext(false); + consumerService.handleGetMessageResult(context, foundResult(110L), + topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100L); + + consumerService.commitPendingOffset(context); + // the offset store caught up after the first commit + Mockito.when(consumerOffsetManager.queryOffset(groupId, topicId, queueId)).thenReturn(110L); + consumerService.commitPendingOffset(context); + + Mockito.verify(consumerOffsetManager, Mockito.times(1)).commitOffset( + anyString(), eq(groupId), eq(topicId), eq(queueId), eq(110L)); + } + + @Test + public void fifoKeepsInlineCommit() { + init(false); + PopConsumerContext context = popContext(true); + consumerService.handleGetMessageResult(context, foundResult(110L), + topicId, queueId, PopConsumerRecord.RetryType.NORMAL_TOPIC, 100L); + + Assert.assertNull(context.getPendingCommitList()); + // fifo commits the batch start offset inline when messages are found + verifyCommitted(100L); + Mockito.verify(consumerOffsetManager, Mockito.never()).commitPullOffset( + anyString(), anyString(), anyString(), anyInt(), anyLong()); + } +} From 3f4340ab5bbadf4565e32a2ef50347b29147a661 Mon Sep 17 00:00:00 2001 From: redlsz Date: Mon, 24 Aug 2026 15:33:48 +0800 Subject: [PATCH 2/4] polish codes --- .../java/org/apache/rocketmq/broker/pop/PopConsumerService.java | 2 +- 1 file changed, 1 insertion(+), 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 0397fde05f2..0e31e2cdb80 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 @@ -484,12 +484,12 @@ protected void commitPendingOffset(PopConsumerContext context) { } String groupId = context.getGroupId(); + ConsumerOffsetManager consumerOffsetManager = this.brokerController.getConsumerOffsetManager(); for (PopConsumerContext.PendingCommit pendingCommit : pendingCommitList) { String topicId = pendingCommit.getTopicId(); int queueId = pendingCommit.getQueueId(); - ConsumerOffsetManager consumerOffsetManager = this.brokerController.getConsumerOffsetManager(); if (consumerOffsetManager.hasOffsetReset(topicId, groupId, queueId)) { continue; } From 2bcd7720d895362570d6b6c05c24f93592324738 Mon Sep 17 00:00:00 2001 From: redlsz Date: Tue, 25 Aug 2026 18:04:48 +0800 Subject: [PATCH 3/4] cleanup timeout records under guard of lock --- .../rocketmq/broker/pop/PopConsumerCache.java | 32 +++++++++++++------ .../broker/pop/PopConsumerCacheTest.java | 16 ++++++++++ 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java index e6740ad29d5..84a5563d5f7 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java @@ -29,6 +29,7 @@ import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.KeyBuilder; import org.apache.rocketmq.common.ServiceThread; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.utils.ConcurrentHashMapUtils; @@ -133,17 +134,28 @@ public int cleanupRecords(Consumer consumer) { records.getGroupId(), records.getTopicId()); if (timeout) { - records.stageExpiredRecords(Long.MAX_VALUE); - List writeConsumerRecords = - new ArrayList<>(records.getRemoveTreeMap().values()); - if (!writeConsumerRecords.isEmpty()) { - consumerRecordStore.writeRecords(writeConsumerRecords); + // hold the same lock as PopConsumerService#popAsync to prevent evicting records that a + // concurrent pop is still writing. + String lockTopicId = KeyBuilder.parseNormalTopic(records.getTopicId(), records.getGroupId()); + if (!consumerLockService.tryLock(records.getGroupId(), lockTopicId)) { + remain += records.getInFlightRecordCount(); + continue; } - records.clearStagedRecords(); - this.commitPendingOffset(records); - log.info("PopConsumerOffline, so clean expire records, groupId={}, topic={}, queueId={}, records={}", - records.getGroupId(), records.getTopicId(), records.getQueueId(), records.getInFlightRecordCount()); - iterator.remove(); + try { + records.stageExpiredRecords(Long.MAX_VALUE); + List writeConsumerRecords = + new ArrayList<>(records.getRemoveTreeMap().values()); + if (!writeConsumerRecords.isEmpty()) { + consumerRecordStore.writeRecords(writeConsumerRecords); + } + records.clearStagedRecords(); + log.info("PopConsumerOffline, so clean expire records, groupId={}, topic={}, queueId={}, records={}", + records.getGroupId(), records.getTopicId(), records.getQueueId(), records.getInFlightRecordCount()); + iterator.remove(); + } finally { + consumerLockService.unlock(records.getGroupId(), lockTopicId); + } + commitPendingOffset(records); continue; } diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java index a9df88a0ab8..335dfe253b4 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java @@ -145,6 +145,22 @@ record = new PopConsumerRecord(2L, groupId, topicId, queueId, consumerRecordList.clear(); consumerCache.cleanupRecords(consumerRecordList::add); Assert.assertEquals(0, consumerRecordList.size()); + + // timeout cleanup is skipped when tryLock fails: concurrent pop holds the lock + record = new PopConsumerRecord(System.currentTimeMillis(), + groupId, topicId, queueId, 0, 20000, 105, attemptId); + consumerCache.writeRecords(Collections.singletonList(record)); + Mockito.when(consumerLockService.tryLock(eq(groupId), eq(topicId))).thenReturn(false); + int remain = consumerCache.cleanupRecords(consumerRecordList::add); + Assert.assertEquals(1, remain); + Assert.assertEquals(1, consumerCache.getCacheKeySize()); + Assert.assertEquals(1, consumerCache.getPopInFlightMessageCount(groupId, topicId, queueId)); + + // timeout cleanup proceeds and persists records to KV store when the lock is acquired + Mockito.when(consumerLockService.tryLock(eq(groupId), eq(topicId))).thenReturn(true); + consumerCache.cleanupRecords(consumerRecordList::add); + Assert.assertEquals(0, consumerCache.getCacheKeySize()); + Mockito.verify(consumerKVStore).writeRecords(Collections.singletonList(record)); } @Test From 59a59235f24a56c3c83407d537ed712441072e7e Mon Sep 17 00:00:00 2001 From: redlsz Date: Tue, 25 Aug 2026 18:56:02 +0800 Subject: [PATCH 4/4] fix log --- .../java/org/apache/rocketmq/broker/pop/PopConsumerCache.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java index 84a5563d5f7..6618e49ea3a 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerCache.java @@ -150,7 +150,7 @@ public int cleanupRecords(Consumer consumer) { } records.clearStagedRecords(); log.info("PopConsumerOffline, so clean expire records, groupId={}, topic={}, queueId={}, records={}", - records.getGroupId(), records.getTopicId(), records.getQueueId(), records.getInFlightRecordCount()); + records.getGroupId(), records.getTopicId(), records.getQueueId(), writeConsumerRecords.size()); iterator.remove(); } finally { consumerLockService.unlock(records.getGroupId(), lockTopicId);