RocketMq retry-topic support SQL filtering messages #1750
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
chengyouling
force-pushed
the
develop-retry
branch
from
0935d59 to
f822209
Compare
Codecov ReportAttention: Patch coverage is
Flags with carried forward coverage won't be shown. Click here to find out more.
|
chengyouling
force-pushed
the
develop-retry
branch
from
5dd3b97 to
f3082d4
Compare
lilai23
reviewed
Feb 24, 2025
...rmant/mq/grayscale/rocketmq/interceptor/RocketMqSchedulerRebuildSubscriptionInterceptor.java
Outdated
Show resolved
Hide resolved
lilai23
reviewed
Feb 24, 2025
Comment on lines
50
to
63
| if (balance.getConsumerGroup() == null || !isGrayTagChanged(map, balance)) { | ||
| return context; | ||
| } | ||
| for (Object subscriptionData : map.values()) { | ||
| String topic = RocketMqReflectUtils.getTopic(subscriptionData); | ||
| if (topic.contains(RETYPE)) { | ||
| continue; | ||
| } | ||
| if (RocketMqSubscriptionDataUtils | ||
| .isExpressionTypeInaccurate(RocketMqReflectUtils.getExpressionType(subscriptionData))) { | ||
| continue; | ||
| } |
There was a problem hiding this comment.
isGrayTagChanged has the similar code with these code. You can traverse the map only once
lilai23
reviewed
Feb 24, 2025
Comment on lines
82
to
100
| private void updateRetrySubscriptionData(Object subscriptionData, Collection<Object> subscriptionDatas) { | ||
| String originTopic = RocketMqReflectUtils.getTopic(subscriptionData); | ||
| for (Object subData : subscriptionDatas) { | ||
| String retryTopic = RocketMqReflectUtils.getTopic(subData); | ||
| String sqlSubstr = RocketMqReflectUtils.getSubString(subscriptionData); | ||
| if (retryTopic.contains(RETYPE)) { | ||
| RocketMqReflectUtils.getTagsSet(subData).clear(); | ||
| RocketMqReflectUtils.getCodeSet(subData).clear(); | ||
| RocketMqReflectUtils.setSubscriptionDatae(subData, "setExpressionType", | ||
| new Class[]{String.class}, new Object[]{"SQL92"}); | ||
| RocketMqReflectUtils.setSubscriptionDatae(subData, "setSubVersion", | ||
| new Class[]{long.class}, new Object[]{System.currentTimeMillis()}); | ||
| String originSubData = RocketMqReflectUtils.getSubString(subData); | ||
| RocketMqReflectUtils.setSubscriptionDatae(subData, "setSubString", | ||
| new Class[]{String.class}, new Object[]{sqlSubstr}); | ||
| LOGGER.warning(String.format(Locale.ENGLISH, "update retry topic [%s] SQL92 expression, " | ||
| + "originTopic: [%s], originSubStr: [%s], newSubStr: [%s]", retryTopic, originTopic, | ||
| originSubData, sqlSubstr)); | ||
| } |
There was a problem hiding this comment.
why traverse map.values() and determine whether the topic contains again here？
chengyouling
force-pushed
the
develop-retry
branch
2 times, most recently
from
b7316e5 to
768cf5c
Compare
daizhenyu
reviewed
Feb 25, 2025
| RocketMqConsumerGroupAutoCheck.setMqClientInstance(topic, consumerGroup, instance); | ||
| RocketMqConsumerGroupAutoCheck.syncUpdateCacheGrayTags(); | ||
| RocketMqConsumerGroupAutoCheck.startSchedulerCheckGroupTask(); | ||
| private boolean isGrayTagChanged(ConcurrentMap<String, Object> map, RebalanceImpl balance, |
There was a problem hiding this comment.
not map，use specific name
chengyouling
force-pushed
the
develop-retry
branch
2 times, most recently
from
cd7a266 to
e0054b4
Compare
lilai23
reviewed
Feb 25, 2025
| for (Object subData : retrySubscriptionDatas) { | ||
| RocketMqReflectUtils.getTagsSet(subData).clear(); | ||
| RocketMqReflectUtils.getCodeSet(subData).clear(); | ||
| RocketMqReflectUtils.setSubscriptionDatae(subData, "setExpressionType", |
There was a problem hiding this comment.
rename setSubscriptionDatae
Signed-off-by: chengyouling <[email protected]>
chengyouling
force-pushed
the
develop-retry
branch
from
93f4e8e to
06e8c45
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What type of PR is this?
Bug.
What this PR does / why we need it?
Retry-topic synchronous adjustment change to using SQL92 filter message.
Which issue(s) this PR fixes？
Fixes #1748
Does this PR introduce a user-facing change?
No
Checklist