unbridled-41 opened a new pull request, #10991: URL: https://github.com/apache/rocketmq/pull/10991
<!-- Please make sure the target branch is right. In most case, the target branch should be `develop`. --> ### Which Issue(s) This PR Fixes - Fixes #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. -- 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]
