From cd89d992ab685acc3d6046b77a1f48f7108d05c2 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 30 Aug 2026 01:31:39 +0800 Subject: [PATCH 1/2] [ISSUE #10987] Fix queryMinOffsetInAllGroup deleting consumer offsets from the live offset table --- .../broker/offset/ConsumerOffsetManager.java | 18 +++++----- .../offset/ConsumerOffsetManagerTest.java | 36 +++++++++++++++++++ 2 files changed, 45 insertions(+), 9 deletions(-) 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 1d3bf7bed09..1facfb5ac30 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 @@ -327,21 +327,21 @@ public ConcurrentMap> getPullOffsetTable() public Map queryMinOffsetInAllGroup(final String topic, final String filterGroups) { Map queueMinOffset = new HashMap<>(); - Set topicGroups = this.offsetTable.keySet(); + // Work on a snapshot of the keys: offsetTable must never be mutated by a query. + Set topicGroups = new HashSet<>(this.offsetTable.keySet()); if (!UtilAll.isBlank(filterGroups)) { for (String group : filterGroups.split(",")) { - Iterator it = topicGroups.iterator(); - while (it.hasNext()) { - String topicAtGroup = it.next(); - if (group.equals(topicAtGroup.split(TOPIC_GROUP_SEPARATOR)[1])) { - it.remove(); - removeConsumerOffset(topicAtGroup); - } - } + topicGroups.removeIf(topicAtGroup -> { + String[] arrays = topicAtGroup.split(TOPIC_GROUP_SEPARATOR); + return arrays.length == 2 && group.equals(arrays[1]); + }); } } for (Map.Entry> offSetEntry : this.offsetTable.entrySet()) { + if (!topicGroups.contains(offSetEntry.getKey())) { + continue; + } String topicGroup = offSetEntry.getKey(); String[] topicGroupArr = topicGroup.split(TOPIC_GROUP_SEPARATOR); if (topic.equals(topicGroupArr[0])) { 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 7e4faa4e42f..c814e8c4428 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 @@ -19,11 +19,13 @@ import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.store.MessageStore; import org.apache.rocketmq.store.config.MessageStoreConfig; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import org.mockito.Mockito; @@ -121,4 +123,38 @@ public void testEraseResetOffset() { Assert.assertFalse(consumerOffsetManager.hasOffsetReset(topic, group, 1)); Assert.assertFalse(consumerOffsetManager.resetOffsetTable.containsKey(key)); } + + @Test + public void testQueryMinOffsetInAllGroupDoesNotDeleteOffsets() { + Mockito.when(brokerController.getBrokerConfig()).thenReturn(new BrokerConfig()); + MessageStore messageStore = Mockito.mock(MessageStore.class); + Mockito.when(brokerController.getMessageStore()).thenReturn(messageStore); + Mockito.when(messageStore.getMinOffsetInQueue(Mockito.anyString(), Mockito.anyInt())).thenReturn(0L); + + String topic = "Topic"; + String group1 = "G1"; + String group2 = "G2"; + ConcurrentHashMap offsets1 = new ConcurrentHashMap<>(); + offsets1.put(0, 50L); + ConcurrentHashMap offsets2 = new ConcurrentHashMap<>(); + offsets2.put(0, 30L); + ConcurrentHashMap> offsetTable = new ConcurrentHashMap<>(); + offsetTable.put(topic + TOPIC_GROUP_SEPARATOR + group1, offsets1); + offsetTable.put(topic + TOPIC_GROUP_SEPARATOR + group2, offsets2); + // malformed key without '@' must not break the query either + offsetTable.put("MalformedKey", new ConcurrentHashMap<>()); + consumerOffsetManager.setOffsetTable(offsetTable); + + // filtering out G2 must exclude its offsets from the min computation + Map result = consumerOffsetManager.queryMinOffsetInAllGroup(topic, group2); + assertThat(result).containsEntry(0, 50L); + + // but the query must not destroy the filtered group's offsets + assertThat(offsetTable).containsKey(topic + TOPIC_GROUP_SEPARATOR + group2); + assertThat(consumerOffsetManager.queryOffset(group2, topic, 0)).isEqualTo(30L); + + // without filter, the min across all groups is returned + result = consumerOffsetManager.queryMinOffsetInAllGroup(topic, ""); + assertThat(result).containsEntry(0, 30L); + } } From 906bab3c2bee9dce65b6c9590f379a1be34c755f Mon Sep 17 00:00:00 2001 From: root Date: Sun, 30 Aug 2026 17:55:57 +0800 Subject: [PATCH 2/2] [ISSUE #10987] Split the queryMinOffsetInAllGroup regression tests so offset deletion and the malformed-key AIOOBE fail independently --- .../offset/ConsumerOffsetManagerTest.java | 20 +++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) 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 c814e8c4428..ae11a0396b6 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 @@ -141,8 +141,6 @@ public void testQueryMinOffsetInAllGroupDoesNotDeleteOffsets() { ConcurrentHashMap> offsetTable = new ConcurrentHashMap<>(); offsetTable.put(topic + TOPIC_GROUP_SEPARATOR + group1, offsets1); offsetTable.put(topic + TOPIC_GROUP_SEPARATOR + group2, offsets2); - // malformed key without '@' must not break the query either - offsetTable.put("MalformedKey", new ConcurrentHashMap<>()); consumerOffsetManager.setOffsetTable(offsetTable); // filtering out G2 must exclude its offsets from the min computation @@ -157,4 +155,22 @@ public void testQueryMinOffsetInAllGroupDoesNotDeleteOffsets() { result = consumerOffsetManager.queryMinOffsetInAllGroup(topic, ""); assertThat(result).containsEntry(0, 30L); } + + @Test + public void testQueryMinOffsetInAllGroupToleratesMalformedKeys() { + Mockito.when(brokerController.getBrokerConfig()).thenReturn(new BrokerConfig()); + MessageStore messageStore = Mockito.mock(MessageStore.class); + Mockito.when(brokerController.getMessageStore()).thenReturn(messageStore); + Mockito.when(messageStore.getMinOffsetInQueue(Mockito.anyString(), Mockito.anyInt())).thenReturn(0L); + + String topic = "Topic"; + ConcurrentHashMap> offsetTable = new ConcurrentHashMap<>(); + offsetTable.put(topic + TOPIC_GROUP_SEPARATOR + "G1", new ConcurrentHashMap<>()); + // malformed key without '@' must not break the query + offsetTable.put("MalformedKey", new ConcurrentHashMap<>()); + consumerOffsetManager.setOffsetTable(offsetTable); + + assertThat(consumerOffsetManager.queryMinOffsetInAllGroup(topic, "G1")).isEmpty(); + assertThat(consumerOffsetManager.queryMinOffsetInAllGroup(topic, "")).isEmpty(); + } }