You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2026/06/29 07:34:06 UTC
(rocketmq) branch develop updated: [ISSUE #10549] Fix lite topic reset offset: memory leak, FIFO block bypass, and offset-0 reset failure (#10550)
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 49102a9985 [ISSUE #10549] Fix lite topic reset offset: memory leak, FIFO block bypass, and offset-0 reset failure (#10550)
49102a9985 is described below
commit 49102a99852147c2fa90d49250f4ffb0d288f3c8
Author: Quan <zs...@foxmail.com>
AuthorDate: Mon Jun 29 15:34:00 2026 +0800
[ISSUE #10549] Fix lite topic reset offset: memory leak, FIFO block bypass, and offset-0 reset failure (#10550)
* [ISSUE #10549] Fix lite topic reset offset: memory leak, FIFO block bypass, and offset-0 reset failure
- Fix memory leak in removeResetOffset: clean up empty inner map entries from resetOffsetTable
- Add eraseResetOffset for precise cleanup on lite topic removal
- Skip FIFO block check in isFifoBlocked when server-side reset offset is pending
- Fix ResetOffsetByTimeCommand: change resetOffset > 0 to >= 0 to allow resetting to offset 0
- Add unit tests for eraseResetOffset and isFifoBlocked reset bypass
* chore: empty commit to trigger CI pipeline
---
.../broker/lite/AbstractLiteLifecycleManager.java | 1 +
.../broker/offset/ConsumerOffsetManager.java | 17 +++++++++++++++--
.../broker/processor/PopLiteMessageProcessor.java | 4 ++++
.../broker/offset/ConsumerOffsetManagerTest.java | 21 +++++++++++++++++++++
.../processor/PopLiteMessageProcessorTest.java | 11 +++++++++++
.../command/offset/ResetOffsetByTimeCommand.java | 2 +-
6 files changed, 53 insertions(+), 3 deletions(-)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
index b038a692d3..f7f522b833 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
@@ -205,6 +205,7 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
groups.forEach(group -> {
String topicAtGroup = lmqName + TOPIC_GROUP_SEPARATOR + group;
brokerController.getConsumerOffsetManager().getOffsetTable().remove(topicAtGroup);
+ brokerController.getConsumerOffsetManager().eraseResetOffset(lmqName, group, 0);
brokerController.getConsumerOffsetManager().removeConsumerOffset(topicAtGroup); // no iteration
brokerController.getPopLiteMessageProcessor().getConsumerOrderInfoManager().remove(lmqName, group);
});
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java b/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
index f9debf3857..1d3bf7bed0 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
@@ -466,8 +466,21 @@ public class ConsumerOffsetManager extends ConfigManager {
ConcurrentMap<Integer, Long> map = resetOffsetTable.get(key);
if (null == map) {
return null;
- } else {
- return map.remove(queueId);
}
+ Long offset = map.remove(queueId);
+ if (map.isEmpty()) {
+ resetOffsetTable.computeIfPresent(key, (k, _map) ->
+ _map.isEmpty() ? null : _map
+ );
+ }
+ return offset;
+ }
+
+ public void eraseResetOffset(String topic, String group, int queueId) {
+ String key = topic + TOPIC_GROUP_SEPARATOR + group;
+ resetOffsetTable.computeIfPresent(key, (k, map) -> {
+ map.remove(queueId);
+ return map.isEmpty() ? null : map;
+ });
}
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java
index a1fa417152..4da72ef4bd 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java
@@ -311,6 +311,10 @@ public class PopLiteMessageProcessor implements NettyRequestProcessor {
}
public boolean isFifoBlocked(String attemptId, String group, String lmqName, long invisibleTime) {
+ if (brokerController.getBrokerConfig().isUseServerSideResetOffset() &&
+ this.brokerController.getConsumerOffsetManager().hasOffsetReset(lmqName, group, 0)) {
+ return false;
+ }
return consumerOrderInfoManager.checkBlock(attemptId, lmqName, group, 0, invisibleTime);
}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java
index 3ddd369c7f..7e4faa4e42 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java
@@ -100,4 +100,25 @@ public class ConsumerOffsetManagerTest {
ConcurrentMap<Integer, Long> offsetTableLoaded = manager.getOffsetTable().get(group);
Assert.assertEquals(table, offsetTableLoaded);
}
+
+ @Test
+ public void testEraseResetOffset() {
+ String topic = "Topic";
+ String group = "Group";
+ String key = topic + TOPIC_GROUP_SEPARATOR + group;
+ consumerOffsetManager.assignResetOffset(topic, group, 0, 100L);
+ consumerOffsetManager.assignResetOffset(topic, group, 1, 200L);
+
+ Assert.assertTrue(consumerOffsetManager.hasOffsetReset(topic, group, 0));
+ Assert.assertTrue(consumerOffsetManager.hasOffsetReset(topic, group, 1));
+
+ consumerOffsetManager.eraseResetOffset(topic, group, 0);
+ Assert.assertFalse(consumerOffsetManager.hasOffsetReset(topic, group, 0));
+ Assert.assertTrue(consumerOffsetManager.hasOffsetReset(topic, group, 1));
+ Assert.assertTrue(consumerOffsetManager.resetOffsetTable.containsKey(key));
+
+ consumerOffsetManager.eraseResetOffset(topic, group, 1);
+ Assert.assertFalse(consumerOffsetManager.hasOffsetReset(topic, group, 1));
+ Assert.assertFalse(consumerOffsetManager.resetOffsetTable.containsKey(key));
+ }
}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java
index 9705ab4f5a..957386b166 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java
@@ -56,6 +56,7 @@ import java.util.Iterator;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -149,6 +150,16 @@ public class PopLiteMessageProcessorTest {
verify(consumerOrderInfoManager).checkBlock("attemptId", "lmqName", "group", 0, 1000L);
}
+ @Test
+ public void testIsFifoBlocked_hasResetOffset() {
+ brokerConfig.setUseServerSideResetOffset(true);
+ when(consumerOffsetManager.hasOffsetReset("lmqName", "group", 0)).thenReturn(true);
+
+ assertFalse(popLiteMessageProcessor.isFifoBlocked("attemptId", "group", "lmqName", 1000L));
+ verify(consumerOffsetManager).hasOffsetReset("lmqName", "group", 0);
+ verify(consumerOrderInfoManager, never()).checkBlock(anyString(), anyString(), anyString(), anyInt(), anyLong());
+ }
+
@Test
public void testGetPopOffset_normal() throws ConsumeQueueException {
String group = "group";
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/offset/ResetOffsetByTimeCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/offset/ResetOffsetByTimeCommand.java
index 84a301bd60..f0326fdb75 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/offset/ResetOffsetByTimeCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/offset/ResetOffsetByTimeCommand.java
@@ -140,7 +140,7 @@ public class ResetOffsetByTimeCommand implements SubCommand {
defaultMQAdminExt.searchOffset(brokerAddr, topic, queueId, timestamp, 3000);
System.out.printf("reset consumer offset to %d%n", resetOffset);
- if (resetOffset > 0) {
+ if (resetOffset >= 0) {
defaultMQAdminExt.resetOffsetByQueueId(brokerAddr, group, topic, queueId, resetOffset);
}
return;