unbridled-41 opened a new issue, #10987: URL: https://github.com/apache/rocketmq/issues/10987
### Before Creating the Bug Report - [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions). - [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe this is not a duplicate. - [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Runtime platform environment - OS: Linux - Component: Broker (`ConsumerOffsetManager`) ### RocketMQ version - branch: develop - Git commit id: e348efa66 ### JDK Version JDK 8 ### Describe the Bug `ConsumerOffsetManager#queryMinOffsetInAllGroup(topic, filterGroups)` iterates the **live** `offsetTable.keySet()` and removes the filtered groups' entries from it: ```java Set<String> topicGroups = this.offsetTable.keySet(); // live view of offsetTable if (!UtilAll.isBlank(filterGroups)) { for (String group : filterGroups.split(",")) { Iterator<String> it = topicGroups.iterator(); while (it.hasNext()) { String topicAtGroup = it.next(); if (group.equals(topicAtGroup.split(TOPIC_GROUP_SEPARATOR)[1])) { it.remove(); // deletes the entry from offsetTable itself removeConsumerOffset(topicAtGroup); } } } } ``` `ConcurrentHashMap.keySet()` is a live view, so `it.remove()` deletes every `topic@group` entry of the filtered groups from the real offset table. This method is called by `AdminBrokerProcessor#queryCorrectionOffset` (`RequestCode.QUERY_CORRECTION_OFFSET`), i.e. by running the read-only admin/diagnostic operation `DefaultMQAdminExt#queryCorrectionOffset(topic, compareGroup, filterGroups)`. Consequences: 1. All consumer offsets of the filtered groups (for every topic on this broker) are wiped from memory; the next `persist()` makes the deletion permanent (`consumers.json` no longer contains the keys). With `RocksDBConsumerOffsetManager` the `removeConsumerOffset` hook deletes the rows from RocksDB immediately. 2. When the consumers of the filtered group commit/look up their offsets afterwards, `queryOffset` returns `-1` and consumption restarts according to `consumeFromWhere` — mass duplicate consumption or consumption skipping to the latest offset. 3. `topicAtGroup.split(TOPIC_GROUP_SEPARATOR)[1]` also throws `ArrayIndexOutOfBoundsException` if a malformed key without `@` is present. This looks like a copy-paste from the real cleanup methods (`cleanOffset` / `removeOffset`): a *query* method must never mutate the table; the filter groups were only meant to be excluded from the min-offset computation. ### Steps to Reproduce 1. Start a broker, let groups `G1` and `G2` consume topic `T` so both groups have committed offsets. 2. Run the admin operation `queryCorrectionOffset(T, G1, filterGroups="G2")` once. 3. Check `consumerOffset.json` / `getConsumerStatus`: every `T@G2` offset entry is gone and gets persisted that way. A unit test asserting `offsetTable` still contains the filtered group's entry after calling `queryMinOffsetInAllGroup` fails on current develop. ### What Did You Expect to See? The query returns the min offsets excluding the filtered groups, without modifying `offsetTable` at all. ### What Did You See Instead? The query deletes the filtered groups' offsets from the live table (and from RocksDB/persisted JSON), causing silent offset loss. ### Additional Context Fix: compute on a snapshot of the key set and exclude the filter groups there, leaving `offsetTable` untouched. I will submit a PR with a regression test. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
