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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
