You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "cherryMJY (via GitHub)" <gi...@apache.org> on 2026/03/20 10:04:04 UTC
[PR] [ISSUE #xxxx] Fix thread-safe issue in ConsumerManager.topicGroupTable [rocketmq]
cherryMJY opened a new pull request, #10179:
URL: https://github.com/apache/rocketmq/pull/10179
Brief Description
The topicGroupTable in ConsumerManager uses HashSet to store consumer groups, which is not thread-safe. When multiple consumers concurrently register with the same topic via heartbeat requests, HashSet.add() may lose entries due to race conditions.
Concurrent scenarios that trigger this bug:
Multiple consumers start up simultaneously and send heartbeat requests
Network reconnect triggers batch consumer re-registration
Proxy syncs consumer info to broker concurrently
Impact:
Consumer groups may not be recorded in topicGroupTable, causing message routing failures
In extreme cases, HashSet internal structure may be corrupted, leading to infinite loops during iteration
Solution:
Replace HashSet with ConcurrentHashMap.newKeySet() which provides thread-safe add/remove operations using CAS + synchronized mechanism. This pattern is already used in other RocketMQ components:
LiteSubscriptionRegistryImpl.liteTopic2Group
TopicList.topicList
LiteSubscription.liteTopicSet
Modified files:
broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #xxxx] Fix thread-safe issue in ConsumerManager.topicGroupTable [rocketmq]
Posted by "lizhimins (via GitHub)" <gi...@apache.org>.
lizhimins commented on code in PR #10179:
URL: https://github.com/apache/rocketmq/pull/10179#discussion_r3013159467
##########
broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java:
##########
@@ -298,9 +300,30 @@ public void check(long transactionTimeout, int transactionCheckMax,
if (isNeedCheck) {
- if (!putBackHalfMsgQueue(msgExt, i)) {
+ int retryTimes = 0;
+ boolean putBackSuccess = false;
+ while (retryTimes < PUT_BACK_RETRY_TIMES) {
Review Comment:
It seems to be no need to introduce retries here.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #xxxx] Fix thread-safe issue in ConsumerManager.topicGroupTable [rocketmq]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #10179:
URL: https://github.com/apache/rocketmq/pull/10179#issuecomment-4102990951
## [Codecov](https://app.codecov.io/gh/apache/rocketmq/pull/10179?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 48.86%. Comparing base ([`ebf1595`](https://app.codecov.io/gh/apache/rocketmq/commit/ebf1595418178d5c73ba80ab08fb3a7312efc25e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`38b71b8`](https://app.codecov.io/gh/apache/rocketmq/commit/38b71b8de0775f737f2a5981f8e7e0f5356cecc3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on develop.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## develop #10179 +/- ##
=============================================
- Coverage 48.94% 48.86% -0.08%
+ Complexity 13389 13367 -22
=============================================
Files 1373 1373
Lines 99924 99920 -4
Branches 12908 12904 -4
=============================================
- Hits 48904 48827 -77
- Misses 45097 45151 +54
- Partials 5923 5942 +19
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/rocketmq/pull/10179?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #xxxx] Fix thread-safe issue in ConsumerManager.topicGroupTable [rocketmq]
Posted by "cherryMJY (via GitHub)" <gi...@apache.org>.
cherryMJY commented on code in PR #10179:
URL: https://github.com/apache/rocketmq/pull/10179#discussion_r3013448806
##########
broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java:
##########
@@ -298,9 +300,30 @@ public void check(long transactionTimeout, int transactionCheckMax,
if (isNeedCheck) {
- if (!putBackHalfMsgQueue(msgExt, i)) {
+ int retryTimes = 0;
+ boolean putBackSuccess = false;
+ while (retryTimes < PUT_BACK_RETRY_TIMES) {
Review Comment:
ok
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #xxxx] Fix thread-safe issue in ConsumerManager.topicGroupTable [rocketmq]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on code in PR #10179:
URL: https://github.com/apache/rocketmq/pull/10179#discussion_r4054257546
##########
broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java:
##########
@@ -298,9 +300,30 @@ public void check(long transactionTimeout, int transactionCheckMax,
if (isNeedCheck) {
- if (!putBackHalfMsgQueue(msgExt, i)) {
+ int retryTimes = 0;
+ boolean putBackSuccess = false;
+ while (retryTimes < PUT_BACK_RETRY_TIMES) {
+ putBackSuccess = putBackHalfMsgQueue(msgExt, i);
+ if (putBackSuccess) {
+ break;
+ }
+ retryTimes++;
+ if (retryTimes < PUT_BACK_RETRY_TIMES) {
+ try {
+ Thread.sleep(100L * retryTimes);
+ } catch (InterruptedException ignored) {
+ }
+ }
Review Comment:
**[Info]** The `catch (InterruptedException e)` block swallows the interrupt without restoring the interrupt status. Consider adding `Thread.currentThread().interrupt()` before the log statement to preserve the interrupt signal for upstream callers.
##########
broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java:
##########
@@ -236,12 +236,10 @@ public boolean registerConsumer(final String group, final ClientChannelInfo clie
}
for (SubscriptionData subscriptionData : subList) {
- Set<String> groups = this.topicGroupTable.get(subscriptionData.getTopic());
- if (groups == null) {
- Set<String> tmp = new HashSet<>();
- Set<String> prev = this.topicGroupTable.putIfAbsent(subscriptionData.getTopic(), tmp);
- groups = prev != null ? prev : tmp;
- }
+ Set<String> groups = this.topicGroupTable.computeIfAbsent(
Review Comment:
**[Info]** Good fix — `ConcurrentHashMap.newKeySet()` is the correct replacement for `HashSet` in concurrent contexts. The use of `computeIfAbsent` also simplifies the code nicely. This pattern is consistent with `LiteSubscriptionRegistryImpl.liteTopic2Group` and `TopicList.topicList` as noted in the PR description.
##########
broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java:
##########
@@ -298,9 +300,30 @@ public void check(long transactionTimeout, int transactionCheckMax,
if (isNeedCheck) {
- if (!putBackHalfMsgQueue(msgExt, i)) {
+ int retryTimes = 0;
+ boolean putBackSuccess = false;
+ while (retryTimes < PUT_BACK_RETRY_TIMES) {
Review Comment:
**[Warning]** The retry logic for `putBackHalfMsgQueue` was questioned by @lizhimins ("It seems to be no need to introduce retries here") and you replied "ok", but the retry code is still present in this PR.
Please either:
1. Remove the retry logic as suggested, or
2. Explain why retries are necessary with data/evidence (e.g., observed transient failures in production)
Leaving unresolved reviewer feedback without action may delay merge.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org