You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2020/07/16 14:59:07 UTC
[GitHub] [rocketmq] jack870131 opened a new pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
jack870131 opened a new pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169
## What is the purpose of the change
XXXXX
## Brief changelog
XX
## Verifying this change
XXXX
Follow this checklist to help us incorporate your contribution quickly and easily. Notice, `it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR`.
- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq/issues) filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
- [x] Format the pull request title like `[ISSUE #123] Fix UnknownException when host config not exist`. Each commit in the pull request should have a meaningful subject line and body.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in [test module](https://github.com/apache/rocketmq/tree/master/test).
- [x] Run `mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle` to make sure basic checks pass. Run `mvn clean install -DskipITs` to make sure unit-test pass. Run `mvn clean test-compile failsafe:integration-test` to make sure integration-test pass.
- [ ] If this contribution is large, please file an [Apache Individual Contributor License Agreement](http://www.apache.org/licenses/#clas).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] RongtongJin commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
RongtongJin commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r456289024
##########
File path: client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java
##########
@@ -278,11 +278,20 @@ private void rebalanceByTopic(final String topic, final boolean isOrder) {
List<MessageQueue> allocateResult = null;
try {
- allocateResult = strategy.allocate(
- this.consumerGroup,
- this.mQClientFactory.getClientId(),
- mqAll,
- cidAll);
+ if (!isRebalanceByBroker()) {
+ allocateResult = strategy.allocate(
+ this.consumerGroup,
+ this.mQClientFactory.getClientId(),
+ mqAll,
+ cidAll);
Review comment:
IMO, It would be better that the client only needs to bring the current cid to get the result.
Both mqAll and cidAll are obtained on the broker side.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] RongtongJin commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
RongtongJin commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r470512389
##########
File path: common/src/main/java/org/apache/rocketmq/common/rebalance/AllocateMessageQueueSticky.java
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.rebalance;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+
+public class AllocateMessageQueueSticky implements AllocateMessageQueueStrategy {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
+ private final Map<String, List<MessageQueue>> messageQueueAllocation;
+
+ private final List<MessageQueue> unassignedQueues = new ArrayList<MessageQueue>();
+
+ public AllocateMessageQueueSticky(Map<String, List<MessageQueue>> messageQueueAllocation) {
+ this.messageQueueAllocation = messageQueueAllocation;
+ }
Review comment:
Why not just new a hashmap in the constructor instead of passing one ？
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32791161)
Coverage decreased (-0.2%) to 51.019% when pulling **91fa427345462eac7fb0c727d32d95e6ff0950cd on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32276646)
Coverage decreased (-0.2%) to 51.025% when pulling **9c4ddc2b8e6bedbe63ea09e82df0dc8004ab27cf on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r459443299
##########
File path: client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java
##########
@@ -278,11 +278,20 @@ private void rebalanceByTopic(final String topic, final boolean isOrder) {
List<MessageQueue> allocateResult = null;
try {
- allocateResult = strategy.allocate(
- this.consumerGroup,
- this.mQClientFactory.getClientId(),
- mqAll,
- cidAll);
+ if (!isRebalanceByBroker()) {
+ allocateResult = strategy.allocate(
+ this.consumerGroup,
+ this.mQClientFactory.getClientId(),
+ mqAll,
+ cidAll);
Review comment:
Updated. According to our previous discussion, the performance of obtaining mqAll from broker side still need further discussions.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32931993)
Coverage decreased (-0.3%) to 50.893% when pulling **87bae705b3502b155bae6f76add21e14a1991f5e on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32751840)
Coverage decreased (-0.3%) to 50.928% when pulling **0a73b304b25d7fc67cdbddd48f7624c825de774a on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls commented on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls commented on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32132174)
Coverage decreased (-0.4%) to 50.877% when pulling **3059642788f063fd41a5ee445483ee668b2b63b4 on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32645374)
Coverage decreased (-0.4%) to 50.876% when pulling **96b7a0055f22ac6ff36aef7bf4c3d1037e10cc17 on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] duhenglucky commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
duhenglucky commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r456218770
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/processor/AllocateMessageQueueProcessor.java
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.processor;
+
+import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.AllocateMessageQueueRequestBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueRequestHeader;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseHeader;
+import org.apache.rocketmq.common.rebalance.AllocateMachineRoomNearby;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragely;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragelyByCircle;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByConfig;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByMachineRoom;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueConsistentHash;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public class AllocateMessageQueueProcessor implements NettyRequestProcessor {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+
+ private final BrokerController brokerController;
+
+ public AllocateMessageQueueProcessor(final BrokerController brokerController) {
+ this.brokerController = brokerController;
+ }
+
+ @Override
+ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ switch (request.getCode()) {
+ case RequestCode.ALLOCATE_MESSAGE_QUEUE:
+ return this.allocateMessageQueue(ctx, request);
+ default:
+ break;
+ }
+ return null;
+ }
+
+ private RemotingCommand allocateMessageQueue(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ final RemotingCommand response =
+ RemotingCommand.createResponseCommand(AllocateMessageQueueResponseHeader.class);
+ final AllocateMessageQueueRequestHeader requestHeader =
+ (AllocateMessageQueueRequestHeader) request.decodeCommandCustomHeader(AllocateMessageQueueRequestHeader.class);
+ final AllocateMessageQueueRequestBody requestBody = AllocateMessageQueueRequestBody.decode(request.getBody(),
+ AllocateMessageQueueRequestBody.class);
+
+ AllocateMessageQueueStrategy strategy = null;
+ String strategyName = requestHeader.getStrategyName();
Review comment:
Recommend to add strategy consistency check logic
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/processor/AllocateMessageQueueProcessor.java
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.processor;
+
+import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.AllocateMessageQueueRequestBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueRequestHeader;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseHeader;
+import org.apache.rocketmq.common.rebalance.AllocateMachineRoomNearby;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragely;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragelyByCircle;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByConfig;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByMachineRoom;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueConsistentHash;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public class AllocateMessageQueueProcessor implements NettyRequestProcessor {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+
+ private final BrokerController brokerController;
+
+ public AllocateMessageQueueProcessor(final BrokerController brokerController) {
+ this.brokerController = brokerController;
+ }
+
+ @Override
+ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ switch (request.getCode()) {
+ case RequestCode.ALLOCATE_MESSAGE_QUEUE:
+ return this.allocateMessageQueue(ctx, request);
+ default:
+ break;
+ }
+ return null;
+ }
+
+ private RemotingCommand allocateMessageQueue(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ final RemotingCommand response =
+ RemotingCommand.createResponseCommand(AllocateMessageQueueResponseHeader.class);
+ final AllocateMessageQueueRequestHeader requestHeader =
+ (AllocateMessageQueueRequestHeader) request.decodeCommandCustomHeader(AllocateMessageQueueRequestHeader.class);
+ final AllocateMessageQueueRequestBody requestBody = AllocateMessageQueueRequestBody.decode(request.getBody(),
+ AllocateMessageQueueRequestBody.class);
+
+ AllocateMessageQueueStrategy strategy = null;
+ String strategyName = requestHeader.getStrategyName();
+ if (strategyName.startsWith("MACHINE_ROOM_NEARBY")) {
+ strategy = new AllocateMachineRoomNearby(new AllocateMessageQueueAveragely(),
+ new AllocateMachineRoomNearby.MachineRoomResolver() {
+ @Override
+ public String brokerDeployIn(MessageQueue messageQueue) {
+ return messageQueue.getBrokerName().split("-")[0];
+ }
+
+ @Override
+ public String consumerDeployIn(String clientID) {
+ return clientID.split("-")[0];
+ }
+ });
+ } else {
+ switch (strategyName) {
+ case "AVG":
Review comment:
Enum or constant would be better，magic numbers not a good choice.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32165028)
Coverage decreased (-0.4%) to 50.802% when pulling **e7d2a99a5c341637bb524fa12279dc47abb832ec on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r471219286
##########
File path: common/src/main/java/org/apache/rocketmq/common/rebalance/AllocateMessageQueueSticky.java
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.rebalance;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+
+public class AllocateMessageQueueSticky implements AllocateMessageQueueStrategy {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
+ private final Map<String, List<MessageQueue>> messageQueueAllocation;
+
+ private final List<MessageQueue> unassignedQueues = new ArrayList<MessageQueue>();
+
+ public AllocateMessageQueueSticky(Map<String, List<MessageQueue>> messageQueueAllocation) {
+ this.messageQueueAllocation = messageQueueAllocation;
+ }
Review comment:
> IMO, It's would be better not to upload LitePullConsumerRebalanceByBrokerReliabilityTest.java and PushConsumerRebalanceByBrokerReliabilityTest.java, because they are not examples and will mislead users.
You're right. Those two tests had been taken off from the commit.
> Why not just new a hashmap in the constructor instead of passing one ？
Fixed. It's just a careless mistake. Thx for pointing out.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r470074684
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java
##########
@@ -152,4 +168,75 @@ private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingC
return response;
}
+
+ private RemotingCommand allocateMessageQueue(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ final RemotingCommand response =
+ RemotingCommand.createResponseCommand(AllocateMessageQueueResponseHeader.class);
+ final AllocateMessageQueueRequestHeader requestHeader =
+ (AllocateMessageQueueRequestHeader) request.decodeCommandCustomHeader(AllocateMessageQueueRequestHeader.class);
+ final AllocateMessageQueueRequestBody requestBody = AllocateMessageQueueRequestBody.decode(request.getBody(),
+ AllocateMessageQueueRequestBody.class);
+
+ AllocateMessageQueueStrategy strategy = null;
+ String consumerGroup = requestHeader.getConsumerGroup();
+ String strategyName = requestHeader.getStrategyName();
+
+ if (this.brokerController.getAllocateMessageQueueStrategyTable().containsKey(consumerGroup)) {
+ strategy = this.brokerController.getAllocateMessageQueueStrategyTable().get(consumerGroup);
+ } else {
+ if (strategyName.startsWith(AllocateMessageQueueStrategyConstants.ALLOCATE_MACHINE_ROOM_NEARBY)) {
+ response.setCode(ResponseCode.ALLOCATE_MESSAGE_QUEUE_STRATEGY_NOT_SUPPORTED);
+ response.setRemark("The broker does not support message queue strategy " + strategyName);
+ return response;
+ } else {
+ switch (strategyName) {
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_AVERAGELY:
+ strategy = new AllocateMessageQueueAveragely();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_AVERAGELY_BY_CIRCLE:
+ strategy = new AllocateMessageQueueAveragelyByCircle();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_BY_CONFIG:
+ strategy = new AllocateMessageQueueByConfig();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_BY_MACHINE_ROOM:
+ strategy = new AllocateMessageQueueByMachineRoom();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_CONSISTENT_HASH:
+ strategy = new AllocateMessageQueueConsistentHash();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_STICKY:
+ strategy = new AllocateMessageQueueSticky(new HashMap<String, List<MessageQueue>>());
+ default:
+ break;
+ }
+ }
+ this.brokerController.getAllocateMessageQueueStrategyTable().put(consumerGroup, strategy);
+ }
Review comment:
Updated. A synchronized keyword had been applied to avoid race conditions.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] xiangwangcheng commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
xiangwangcheng commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r456178644
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/processor/AllocateMessageQueueProcessor.java
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.processor;
+
+import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.AllocateMessageQueueRequestBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueRequestHeader;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseHeader;
+import org.apache.rocketmq.common.rebalance.AllocateMachineRoomNearby;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragely;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragelyByCircle;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByConfig;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByMachineRoom;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueConsistentHash;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public class AllocateMessageQueueProcessor implements NettyRequestProcessor {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+
+ private final BrokerController brokerController;
+
+ public AllocateMessageQueueProcessor(final BrokerController brokerController) {
+ this.brokerController = brokerController;
+ }
+
+ @Override
+ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ switch (request.getCode()) {
+ case RequestCode.ALLOCATE_MESSAGE_QUEUE:
+ return this.allocateMessageQueue(ctx, request);
+ default:
+ break;
+ }
+ return null;
+ }
+
+ private RemotingCommand allocateMessageQueue(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ final RemotingCommand response =
+ RemotingCommand.createResponseCommand(AllocateMessageQueueResponseHeader.class);
+ final AllocateMessageQueueRequestHeader requestHeader =
+ (AllocateMessageQueueRequestHeader) request.decodeCommandCustomHeader(AllocateMessageQueueRequestHeader.class);
+ final AllocateMessageQueueRequestBody requestBody = AllocateMessageQueueRequestBody.decode(request.getBody(),
+ AllocateMessageQueueRequestBody.class);
+
+ AllocateMessageQueueStrategy strategy = null;
+ String strategyName = requestHeader.getStrategyName();
+ if (strategyName.startsWith("MACHINE_ROOM_NEARBY")) {
+ strategy = new AllocateMachineRoomNearby(new AllocateMessageQueueAveragely(),
Review comment:
Would it be better if we cache the commonly used Strategy Object on the broker side? I think it's better than creating it for every request.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32753330)
Coverage decreased (-0.3%) to 50.957% when pulling **3549cac6f4b00ca6d4ae3f23e7609011ae582cd9 on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] codecov-commenter commented on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-663024854
# [Codecov](https://codecov.io/gh/apache/rocketmq/pull/2169?src=pr&el=h1) Report
> Merging [#2169](https://codecov.io/gh/apache/rocketmq/pull/2169?src=pr&el=desc) into [optimize_rebalance](https://codecov.io/gh/apache/rocketmq/commit/fac30c3596d5577cd80a52af9ad89a0839016137&el=desc) will **decrease** coverage by `0.38%`.
> The diff coverage is `35.71%`.
[](https://codecov.io/gh/apache/rocketmq/pull/2169?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## optimize_rebalance #2169 +/- ##
========================================================
- Coverage 45.64% 45.25% -0.39%
+ Complexity 4265 4239 -26
========================================================
Files 546 551 +5
Lines 35860 35987 +127
Branches 4763 4774 +11
========================================================
- Hits 16369 16287 -82
- Misses 17445 17673 +228
+ Partials 2046 2027 -19
```
| [Impacted Files](https://codecov.io/gh/apache/rocketmq/pull/2169?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...g/apache/rocketmq/client/impl/MQClientAPIImpl.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9NUUNsaWVudEFQSUltcGwuamF2YQ==) | `12.43% <0.00%> (-0.21%)` | `25.00 <0.00> (ø)` | |
| [...ent/impl/consumer/DefaultLitePullConsumerImpl.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9jb25zdW1lci9EZWZhdWx0TGl0ZVB1bGxDb25zdW1lckltcGwuamF2YQ==) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (ø)` | |
| [...lient/impl/consumer/DefaultMQPullConsumerImpl.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9jb25zdW1lci9EZWZhdWx0TVFQdWxsQ29uc3VtZXJJbXBsLmphdmE=) | `27.29% <0.00%> (-0.08%)` | `18.00 <0.00> (ø)` | |
| [...cketmq/client/impl/consumer/RebalancePullImpl.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvaW1wbC9jb25zdW1lci9SZWJhbGFuY2VQdWxsSW1wbC5qYXZh) | `23.80% <0.00%> (-1.20%)` | `2.00 <0.00> (ø)` | |
| [...g/apache/rocketmq/common/protocol/RequestCode.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jb21tb24vcHJvdG9jb2wvUmVxdWVzdENvZGUuamF2YQ==) | `0.00% <ø> (ø)` | `0.00 <0.00> (ø)` | |
| [.../apache/rocketmq/common/protocol/ResponseCode.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jb21tb24vcHJvdG9jb2wvUmVzcG9uc2VDb2RlLmphdmE=) | `0.00% <ø> (ø)` | `0.00 <0.00> (ø)` | |
| [...protocol/body/AllocateMessageQueueRequestBody.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jb21tb24vcHJvdG9jb2wvYm9keS9BbGxvY2F0ZU1lc3NhZ2VRdWV1ZVJlcXVlc3RCb2R5LmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...ocol/header/AllocateMessageQueueRequestHeader.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jb21tb24vcHJvdG9jb2wvaGVhZGVyL0FsbG9jYXRlTWVzc2FnZVF1ZXVlUmVxdWVzdEhlYWRlci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...tocol/header/AllocateMessageQueueResponseBody.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jb21tb24vcHJvdG9jb2wvaGVhZGVyL0FsbG9jYXRlTWVzc2FnZVF1ZXVlUmVzcG9uc2VCb2R5LmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...col/header/AllocateMessageQueueResponseHeader.java](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jb21tb24vcHJvdG9jb2wvaGVhZGVyL0FsbG9jYXRlTWVzc2FnZVF1ZXVlUmVzcG9uc2VIZWFkZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| ... and [47 more](https://codecov.io/gh/apache/rocketmq/pull/2169/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/rocketmq/pull/2169?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/rocketmq/pull/2169?src=pr&el=footer). Last update [fac30c3...9c4ddc2](https://codecov.io/gh/apache/rocketmq/pull/2169?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32679578)
Coverage decreased (-0.4%) to 50.845% when pulling **e9ef67ad7a22909d9dce84fddde11fb541985f63 on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r459441258
##########
File path: client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
##########
@@ -1161,6 +1161,26 @@ public void resetOffset(String topic, String group, Map<MessageQueue, Long> offs
}
}
+ public List<MessageQueue> getAllocationResult(final String topic, final String group, final String strategyName,
+ final List<MessageQueue> mqAll, final List<String> cidAll) {
+ String brokerAddr = this.findBrokerAddrByTopic(topic);
+ if (null == brokerAddr) {
+ this.updateTopicRouteInfoFromNameServer(topic);
+ brokerAddr = this.findBrokerAddrByTopic(topic);
+ }
+
+ if (null != brokerAddr) {
+ try {
+ return this.mQClientAPIImpl.getAllocationResultByStrategy(brokerAddr, group, clientId, strategyName,
+ mqAll, cidAll, 3000);
+ } catch (Exception e) {
+ log.warn("getAllocationResultByStrategy exception, " + brokerAddr, " " + group, e);
Review comment:
Updated.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r459448797
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
##########
@@ -329,6 +331,10 @@ public boolean initialize() throws CloneNotSupportedException {
Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
"ConsumerManageThread_"));
+ this.allocateMessageQueueExecutor =
+ Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
+ "AllocateMessageQueueThread_"));
Review comment:
Updated. The processor thread had been deleted and the code had migrated to ConsumerManageProcessor.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r470073993
##########
File path: common/pom.xml
##########
@@ -24,6 +24,18 @@
<modelVersion>4.0.0</modelVersion>
<packaging>jar</packaging>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>8</source>
+ <target>8</target>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
Review comment:
Updated. It seems like an automatic update from IDE and I forgot to change it back.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r459441048
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/processor/AllocateMessageQueueProcessor.java
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.processor;
+
+import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.AllocateMessageQueueRequestBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueRequestHeader;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseHeader;
+import org.apache.rocketmq.common.rebalance.AllocateMachineRoomNearby;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragely;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragelyByCircle;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByConfig;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByMachineRoom;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueConsistentHash;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public class AllocateMessageQueueProcessor implements NettyRequestProcessor {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+
+ private final BrokerController brokerController;
+
+ public AllocateMessageQueueProcessor(final BrokerController brokerController) {
+ this.brokerController = brokerController;
+ }
+
+ @Override
+ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ switch (request.getCode()) {
+ case RequestCode.ALLOCATE_MESSAGE_QUEUE:
+ return this.allocateMessageQueue(ctx, request);
+ default:
+ break;
+ }
+ return null;
+ }
+
+ private RemotingCommand allocateMessageQueue(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ final RemotingCommand response =
+ RemotingCommand.createResponseCommand(AllocateMessageQueueResponseHeader.class);
+ final AllocateMessageQueueRequestHeader requestHeader =
+ (AllocateMessageQueueRequestHeader) request.decodeCommandCustomHeader(AllocateMessageQueueRequestHeader.class);
+ final AllocateMessageQueueRequestBody requestBody = AllocateMessageQueueRequestBody.decode(request.getBody(),
+ AllocateMessageQueueRequestBody.class);
+
+ AllocateMessageQueueStrategy strategy = null;
+ String strategyName = requestHeader.getStrategyName();
+ if (strategyName.startsWith("MACHINE_ROOM_NEARBY")) {
+ strategy = new AllocateMachineRoomNearby(new AllocateMessageQueueAveragely(),
Review comment:
Hi, good point. I had created a ConcurrentHashMap to cache Strategy Object.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] coveralls edited a comment on pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#issuecomment-659801212
[](https://coveralls.io/builds/32164706)
Coverage decreased (-0.3%) to 50.883% when pulling **91f6d54139e2efc90dded338adc62d3ee921be29 on jack870131:optimize_rebalance** into **fac30c3596d5577cd80a52af9ad89a0839016137 on apache:optimize_rebalance**.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] RongtongJin merged pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
RongtongJin merged pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] RongtongJin commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
RongtongJin commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r456294662
##########
File path: client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
##########
@@ -1161,6 +1161,26 @@ public void resetOffset(String topic, String group, Map<MessageQueue, Long> offs
}
}
+ public List<MessageQueue> getAllocationResult(final String topic, final String group, final String strategyName,
+ final List<MessageQueue> mqAll, final List<String> cidAll) {
+ String brokerAddr = this.findBrokerAddrByTopic(topic);
+ if (null == brokerAddr) {
+ this.updateTopicRouteInfoFromNameServer(topic);
+ brokerAddr = this.findBrokerAddrByTopic(topic);
+ }
+
+ if (null != brokerAddr) {
+ try {
+ return this.mQClientAPIImpl.getAllocationResultByStrategy(brokerAddr, group, clientId, strategyName,
+ mqAll, cidAll, 3000);
+ } catch (Exception e) {
+ log.warn("getAllocationResultByStrategy exception, " + brokerAddr, " " + group, e);
Review comment:
It would be better to use placeholders, such as `log.warn("getAllocationResultByStrategy exception, {} {}", brokerAddr, group, e);`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] RongtongJin commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
RongtongJin commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r468360403
##########
File path: common/src/main/java/org/apache/rocketmq/common/rebalance/AllocateMessageQueueSticky.java
##########
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.rebalance;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+
+public class AllocateMessageQueueSticky implements AllocateMessageQueueStrategy {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
+ private final Map<String, List<MessageQueue>> messageQueueAllocation;
+
+ private final List<MessageQueue> unassignedQueues = new ArrayList<MessageQueue>();
+
+ public AllocateMessageQueueSticky(Map<String, List<MessageQueue>> messageQueueAllocation)
+ throws NullPointerException {
+ if (messageQueueAllocation == null) {
+ throw new NullPointerException("currentMessageQueueAllocation is null");
+ }
Review comment:
Throwing NPE is very strange, because even if you don’t write related code, it will still throw NPE.
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java
##########
@@ -152,4 +168,75 @@ private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingC
return response;
}
+
+ private RemotingCommand allocateMessageQueue(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ final RemotingCommand response =
+ RemotingCommand.createResponseCommand(AllocateMessageQueueResponseHeader.class);
+ final AllocateMessageQueueRequestHeader requestHeader =
+ (AllocateMessageQueueRequestHeader) request.decodeCommandCustomHeader(AllocateMessageQueueRequestHeader.class);
+ final AllocateMessageQueueRequestBody requestBody = AllocateMessageQueueRequestBody.decode(request.getBody(),
+ AllocateMessageQueueRequestBody.class);
+
+ AllocateMessageQueueStrategy strategy = null;
+ String consumerGroup = requestHeader.getConsumerGroup();
+ String strategyName = requestHeader.getStrategyName();
+
+ if (this.brokerController.getAllocateMessageQueueStrategyTable().containsKey(consumerGroup)) {
+ strategy = this.brokerController.getAllocateMessageQueueStrategyTable().get(consumerGroup);
+ } else {
+ if (strategyName.startsWith(AllocateMessageQueueStrategyConstants.ALLOCATE_MACHINE_ROOM_NEARBY)) {
+ response.setCode(ResponseCode.ALLOCATE_MESSAGE_QUEUE_STRATEGY_NOT_SUPPORTED);
+ response.setRemark("The broker does not support message queue strategy " + strategyName);
+ return response;
+ } else {
+ switch (strategyName) {
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_AVERAGELY:
+ strategy = new AllocateMessageQueueAveragely();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_AVERAGELY_BY_CIRCLE:
+ strategy = new AllocateMessageQueueAveragelyByCircle();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_BY_CONFIG:
+ strategy = new AllocateMessageQueueByConfig();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_BY_MACHINE_ROOM:
+ strategy = new AllocateMessageQueueByMachineRoom();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_CONSISTENT_HASH:
+ strategy = new AllocateMessageQueueConsistentHash();
+ break;
+ case AllocateMessageQueueStrategyConstants.ALLOCATE_MESSAGE_QUEUE_STICKY:
+ strategy = new AllocateMessageQueueSticky(new HashMap<String, List<MessageQueue>>());
+ default:
+ break;
+ }
+ }
+ this.brokerController.getAllocateMessageQueueStrategyTable().put(consumerGroup, strategy);
+ }
Review comment:
There may be a thread-safe issue, because multiple threads may call allocateMessageQueue at the same time
##########
File path: common/pom.xml
##########
@@ -24,6 +24,18 @@
<modelVersion>4.0.0</modelVersion>
<packaging>jar</packaging>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>8</source>
+ <target>8</target>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
Review comment:
Is it necessary to force the compilation version to 8, which may cause some compatibility issues
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] RongtongJin commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
RongtongJin commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r458535303
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
##########
@@ -329,6 +331,10 @@ public boolean initialize() throws CloneNotSupportedException {
Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
"ConsumerManageThread_"));
+ this.allocateMessageQueueExecutor =
+ Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
+ "AllocateMessageQueueThread_"));
Review comment:
IMO, we don’t need to open a thread pool separately, could we reuse consumerManageExecutor?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r470070931
##########
File path: common/src/main/java/org/apache/rocketmq/common/rebalance/AllocateMessageQueueSticky.java
##########
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.rebalance;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+
+public class AllocateMessageQueueSticky implements AllocateMessageQueueStrategy {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
+ private final Map<String, List<MessageQueue>> messageQueueAllocation;
+
+ private final List<MessageQueue> unassignedQueues = new ArrayList<MessageQueue>();
+
+ public AllocateMessageQueueSticky(Map<String, List<MessageQueue>> messageQueueAllocation)
+ throws NullPointerException {
+ if (messageQueueAllocation == null) {
+ throw new NullPointerException("currentMessageQueueAllocation is null");
+ }
Review comment:
Updated. Thx for pointing out.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r459447138
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/processor/AllocateMessageQueueProcessor.java
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.processor;
+
+import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.AllocateMessageQueueRequestBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueRequestHeader;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseHeader;
+import org.apache.rocketmq.common.rebalance.AllocateMachineRoomNearby;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragely;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragelyByCircle;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByConfig;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByMachineRoom;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueConsistentHash;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public class AllocateMessageQueueProcessor implements NettyRequestProcessor {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+
+ private final BrokerController brokerController;
+
+ public AllocateMessageQueueProcessor(final BrokerController brokerController) {
+ this.brokerController = brokerController;
+ }
+
+ @Override
+ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ switch (request.getCode()) {
+ case RequestCode.ALLOCATE_MESSAGE_QUEUE:
+ return this.allocateMessageQueue(ctx, request);
+ default:
+ break;
+ }
+ return null;
+ }
+
+ private RemotingCommand allocateMessageQueue(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ final RemotingCommand response =
+ RemotingCommand.createResponseCommand(AllocateMessageQueueResponseHeader.class);
+ final AllocateMessageQueueRequestHeader requestHeader =
+ (AllocateMessageQueueRequestHeader) request.decodeCommandCustomHeader(AllocateMessageQueueRequestHeader.class);
+ final AllocateMessageQueueRequestBody requestBody = AllocateMessageQueueRequestBody.decode(request.getBody(),
+ AllocateMessageQueueRequestBody.class);
+
+ AllocateMessageQueueStrategy strategy = null;
+ String strategyName = requestHeader.getStrategyName();
+ if (strategyName.startsWith("MACHINE_ROOM_NEARBY")) {
+ strategy = new AllocateMachineRoomNearby(new AllocateMessageQueueAveragely(),
+ new AllocateMachineRoomNearby.MachineRoomResolver() {
+ @Override
+ public String brokerDeployIn(MessageQueue messageQueue) {
+ return messageQueue.getBrokerName().split("-")[0];
+ }
+
+ @Override
+ public String consumerDeployIn(String clientID) {
+ return clientID.split("-")[0];
+ }
+ });
+ } else {
+ switch (strategyName) {
+ case "AVG":
Review comment:
Updated. The strings have been changed to constants.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] jack870131 commented on a change in pull request #2169: [ISSUE #2149] Apache RocketMQ rebalancing architecture optimization
Posted by GitBox <gi...@apache.org>.
jack870131 commented on a change in pull request #2169:
URL: https://github.com/apache/rocketmq/pull/2169#discussion_r459445005
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/processor/AllocateMessageQueueProcessor.java
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.processor;
+
+import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.AllocateMessageQueueStrategy;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.AllocateMessageQueueRequestBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueRequestHeader;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseBody;
+import org.apache.rocketmq.common.protocol.header.AllocateMessageQueueResponseHeader;
+import org.apache.rocketmq.common.rebalance.AllocateMachineRoomNearby;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragely;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueAveragelyByCircle;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByConfig;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueByMachineRoom;
+import org.apache.rocketmq.common.rebalance.AllocateMessageQueueConsistentHash;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public class AllocateMessageQueueProcessor implements NettyRequestProcessor {
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+
+ private final BrokerController brokerController;
+
+ public AllocateMessageQueueProcessor(final BrokerController brokerController) {
+ this.brokerController = brokerController;
+ }
+
+ @Override
+ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ switch (request.getCode()) {
+ case RequestCode.ALLOCATE_MESSAGE_QUEUE:
+ return this.allocateMessageQueue(ctx, request);
+ default:
+ break;
+ }
+ return null;
+ }
+
+ private RemotingCommand allocateMessageQueue(ChannelHandlerContext ctx, RemotingCommand request)
+ throws RemotingCommandException {
+ final RemotingCommand response =
+ RemotingCommand.createResponseCommand(AllocateMessageQueueResponseHeader.class);
+ final AllocateMessageQueueRequestHeader requestHeader =
+ (AllocateMessageQueueRequestHeader) request.decodeCommandCustomHeader(AllocateMessageQueueRequestHeader.class);
+ final AllocateMessageQueueRequestBody requestBody = AllocateMessageQueueRequestBody.decode(request.getBody(),
+ AllocateMessageQueueRequestBody.class);
+
+ AllocateMessageQueueStrategy strategy = null;
+ String strategyName = requestHeader.getStrategyName();
Review comment:
Updated. I had added the try/catch statement for the strategy.allocate() method, which could enhance the consistency 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org