[ISSUE #10987] Fix queryMinOffsetInAllGroup deleting consumer offsets from the live offset table

[unbridled-41]

Which Issue(s) This PR Fixes

• Fixes [Bug] QUERY_CORRECTION_OFFSET admin query permanently deletes the filtered groups' consumer offsets #10987

Brief Description

ConsumerOffsetManager#queryMinOffsetInAllGroup(topic, filterGroups) iterated the live offsetTable.keySet() and called it.remove() on it to exclude the filter groups. Since ConcurrentHashMap.keySet() is a live view, running the read-only admin operation QUERY_CORRECTION_OFFSET (AdminBrokerProcessor#queryCorrectionOffset, exposed via DefaultMQAdminExt#queryCorrectionOffset) permanently deleted every topic@group offset entry of the filtered groups:

• the in-memory offsets are gone and the next persist() makes the deletion permanent (consumerOffset.json);
• with RocksDBConsumerOffsetManager, removeConsumerOffset deletes the rows from RocksDB immediately;
• consumers of the filtered group then see -1 from queryOffset and re-initialize per consumeFromWhere → mass duplicate consumption or skipping to max;
• topicAtGroup.split(TOPIC_GROUP_SEPARATOR)[1] also threw ArrayIndexOutOfBoundsException on a malformed key without @.

This PR makes the exclusion work on a snapshot of the key set, so the query no longer mutates offsetTable at all (and never calls removeConsumerOffset), while preserving the original filter semantics: offsets of the filter groups are excluded from the min-offset computation. The malformed-key AIOOBE is fixed by checking arrays.length == 2.

How Did You Test This Change?

Added ConsumerOffsetManagerTest#testQueryMinOffsetInAllGroupDoesNotDeleteOffsets:

• asserts the filtered group is excluded from the returned min offsets;
• asserts the filtered group's offsets are still in the table (and queryOffset still returns them) after the query;
• includes a malformed key without @ to cover the AIOOBE.

mvn -pl broker test -Dtest=ConsumerOffsetManagerTest passes (6/6). On the unfixed code the new test fails with ArrayIndexOutOfBoundsException / observes the deleted entry.

[RockteMQ-AI]

Summary

This PR fixes a critical bug where queryMinOffsetInAllGroup was mutating the live offsetTable by deleting consumer offsets for filtered groups. The original code iterated over the live keySet and called removeConsumerOffset(), which destroyed offset data. The fix works on a snapshot of keys (new HashSet<>(this.offsetTable.keySet())) and uses removeIf on the snapshot instead. A comprehensive test verifies that filtered group offsets are preserved after the query.

LGTM — excellent fix for this data corruption bug!

---

Automated review by github-manager-bot

[unbridled-41]

Test evidence (before → after)

The regression tests were verified in both directions on JDK 8 (mvn -pl broker test -Dtest=ConsumerOffsetManagerTest):

On the unfixed code (fix reverted, tests kept), the two regression tests now fail independently and each one demonstrates one defect:

1. Data destruction — the filtered group's offsets are gone from the live table after the query:

java.lang.AssertionError:
Expecting actual:
  {"Topic@G1"={0=50L}}
to contain key:
  "Topic@G2"
	at ...ConsumerOffsetManagerTest.testQueryMinOffsetInAllGroupDoesNotDeleteOffsets

Note the query returned while Topic@G2 (30L) was silently deleted from offsetTable; with RocksDBConsumerOffsetManager this deletion is immediately persisted to RocksDB via removeConsumerOffset, and with JSON config it is persisted by the next persist().

2. Malformed key handling — ArrayIndexOutOfBoundsException: Index 1 out of bounds for length 1 from topicAtGroup.split(TOPIC_GROUP_SEPARATOR)[1] on any topic@group key without @.

With this PR: Tests run: 7, Failures: 0, Errors: 0, Skipped: 0 (5 pre-existing + 2 new tests).

I split the originally single test into testQueryMinOffsetInAllGroupDoesNotDeleteOffsets and testQueryMinOffsetInAllGroupToleratesMalformedKeys (commit 906bab3) so that the offset-deletion failure is directly observable — previously the AIOOBE from the malformed key masked the deletion assertion.

Side note on CI: the workflow runs for this PR are in action_required state (first-time contributor), so they will start once a maintainer approves them. Local checkstyle baseline (style/rmq_checkstyle.xml) reports ~14.9k pre-existing violations on develop itself; the change introduces no new violation categories beyond the long-line style already used throughout these files.

[RockteMQ-AI]

Summary

This PR modifies 2 files (108 lines changed).

Automated scan completed. A maintainer should do a detailed review of the logic changes.

Files Changed

• broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
• broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManagerTest.java

---

Automated review by github-manager-bot

[RockteMQ-AI]

Summary

This PR fixes a critical bug where queryMinOffsetInAllGroup was mutating the live offsetTable by removing filtered groups' offsets via Iterator.remove() and removeConsumerOffset(). The fix correctly works on a HashSet snapshot and uses the filtered set as a membership guard in the subsequent iteration — no more side-effects from a read-only query.

The added tests cover both the core regression (filtered group's offsets survive the query) and the edge case of malformed keys without the @ separator.

Looks good. 👍

---

Automated review by github-manager-bot