opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-10-25 10:03:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #9244] Avoid writing dirty data in consumption mode (#9245)

Browse Source
This commit is contained in:
hqbfz
2025-07-19 18:17:56 +08:00
committed by RongtongJin
parent 671a090d16
commit d07a946749
1 changed files with 14 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessor.java
+14
View File
@@ -33,6 +33,7 @@ import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragelyByCircle;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.message.MessageQueueAssignment;
@@ -49,6 +50,7 @@ import org.apache.rocketmq.remoting.protocol.body.QueryAssignmentRequestBody;
import org.apache.rocketmq.remoting.protocol.body.QueryAssignmentResponseBody;
import org.apache.rocketmq.remoting.protocol.body.SetMessageRequestModeRequestBody;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
public class QueryAssignmentProcessor implements NettyRequestProcessor {
private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
@@ -314,8 +316,20 @@ public class QueryAssignmentProcessor implements NettyRequestProcessor {
response.setRemark("retry topic is not allowed to set mode");
return response;
}
TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topic);
if (null == topicConfig) {
response.setCode(ResponseCode.TOPIC_NOT_EXIST);
response.setRemark("topic[" + topic + "] not exist");
return response;
}
final String consumerGroup = requestBody.getConsumerGroup();
SubscriptionGroupConfig groupConfig = this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(consumerGroup);
if (null == groupConfig) {
response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
response.setRemark("subscription group does not exist");
return response;
}
this.messageRequestModeManager.setMessageRequestMode(topic, consumerGroup, requestBody);
this.messageRequestModeManager.persist();