Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[ISSUE #8912] Rebalancing logic shift in POP consumption mode #8923

Open
wants to merge 14 commits into
base: develop
Choose a base branch
from
Open

[ISSUE #8912] Rebalancing logic shift in POP consumption mode #8923

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
a217fe2
Initialization preparation
Nov 13, 2024
b651b20
Optimize client rebalancing logic
Oct 19, 2024
1128571
The server actively notifies
Nov 13, 2024
67ed8f7
Implement a scheduled task
Nov 13, 2024
57be240
The client implements the rebalancing logic in pop consumption mode
Nov 14, 2024
6445e6c
The client implements the rebalancing logic in pop consumption mode
Nov 14, 2024
cbdc87f
The client implements the rebalancing logic in pop consumption mode
Nov 14, 2024
8a94412
The client implements the rebalancing logic in pop consumption mode
Nov 14, 2024
383285f
The client implements the rebalancing logic in pop consumption mode
Nov 14, 2024
e067d7c
Merge branch 'apache:develop' into rebalance
3424672656 Nov 14, 2024
f214dcf
fix test
Nov 14, 2024
3935a22
fix test
Nov 14, 2024
aeb3acd
optimize logic
Nov 15, 2024
4007355
add test
Nov 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.message.MessageQueueForC;
import org.apache.rocketmq.common.message.MessageRequestMode;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
Expand All @@ -46,6 +47,7 @@
import org.apache.rocketmq.remoting.protocol.body.GetConsumerStatusBody;
import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBody;
import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBodyForC;
import org.apache.rocketmq.remoting.protocol.body.SetMessageRequestModeRequestBody;
import org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.GetConsumerStatusRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.NotifyConsumerIdsChangedRequestHeader;
Expand Down Expand Up @@ -82,6 +84,24 @@ public RemotingCommand callClient(final Channel channel,
return this.brokerController.getRemotingServer().invokeSync(channel, request, 10000);
}

public void notifyMessageRequestModeChange(final Channel channel,
final String consumerGroup, final String topic, final MessageRequestMode mode, int popShareQueueNum) {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.NOTIFY_MESSAGE_REQUEST_MODE_TO_CLIENT, null);

SetMessageRequestModeRequestBody setMessageRequestModeRequestBody = new SetMessageRequestModeRequestBody();
setMessageRequestModeRequestBody.setTopic(topic);
setMessageRequestModeRequestBody.setConsumerGroup(consumerGroup);
setMessageRequestModeRequestBody.setPopShareQueueNum(popShareQueueNum);
setMessageRequestModeRequestBody.setMode(mode);
request.setBody(setMessageRequestModeRequestBody.encode());

try {
this.brokerController.getRemotingServer().invokeOneway(channel, request, 10);
} catch (Exception e) {
log.error("notifyMessageRequestMode to client exception. group={}, error={}", consumerGroup, e.toString());
}
}

public void notifyConsumerIdsChanged(
final Channel channel,
final String consumerGroup) {
Expand Down
10 changes: 10 additions & 0 deletions broker/src/main/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.broker.processor;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.util.ArrayList;
import java.util.Collections;
Expand Down Expand Up @@ -316,10 +317,19 @@ private RemotingCommand setMessageRequestMode(ChannelHandlerContext ctx,
}

final String consumerGroup = requestBody.getConsumerGroup();
ConsumerGroupInfo consumerGroupInfo =
this.brokerController.getConsumerManager().getConsumerGroupInfo(consumerGroup);

this.messageRequestModeManager.setMessageRequestMode(topic, consumerGroup, requestBody);
this.messageRequestModeManager.persist();

if (consumerGroupInfo != null) {
for (Channel channel : consumerGroupInfo.getAllChannel()) {
this.brokerController.getBroker2Client().notifyMessageRequestModeChange(channel, consumerGroup, requestBody.getTopic(),
requestBody.getMode(), requestBody.getPopShareQueueNum());
}
}

response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
return response;
Expand Down
11 changes: 11 additions & 0 deletions client/src/main/java/org/apache/rocketmq/client/ClientConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,8 @@ public class ClientConfig {
*/
protected String traceTopic;

private boolean enableRebalanceTransferInPop = false;

public String buildMQClientId() {
StringBuilder sb = new StringBuilder();
sb.append(this.getClientIP());
Expand Down Expand Up @@ -515,6 +517,14 @@ public void setTraceTopic(String traceTopic) {
this.traceTopic = traceTopic;
}

public boolean getEnableRebalanceTransferInPop() {
return enableRebalanceTransferInPop;
}

public void setEnableRebalanceTransferInPop(boolean enableRebalanceTransferInPop) {
this.enableRebalanceTransferInPop = enableRebalanceTransferInPop;
}

@Override
public String toString() {
return "ClientConfig{" +
Expand Down Expand Up @@ -548,6 +558,7 @@ public String toString() {
", enableHeartbeatChannelEventListener=" + enableHeartbeatChannelEventListener +
", enableTrace=" + enableTrace +
", traceTopic='" + traceTopic + '\'' +
", enableRebalanceTransfer=" + enableRebalanceTransferInPop +
'}';
}
}
21 changes: 21 additions & 0 deletions client/src/main/java/org/apache/rocketmq/client/impl/ClientRemotingProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.message.MessageRequestMode;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.remoting.common.RemotingHelper;
Expand All @@ -47,6 +48,7 @@
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.GetConsumerStatusBody;
import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBody;
import org.apache.rocketmq.remoting.protocol.body.SetMessageRequestModeRequestBody;
import org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ConsumeMessageDirectlyResultRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
Expand Down Expand Up @@ -86,6 +88,9 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx,

case RequestCode.PUSH_REPLY_MESSAGE_TO_CLIENT:
return this.receiveReplyMessage(ctx, request);

case RequestCode.NOTIFY_MESSAGE_REQUEST_MODE_TO_CLIENT:
return this.updateMessageRequestMode(ctx, request);
default:
break;
}
Expand Down Expand Up @@ -292,4 +297,20 @@ private void processReplyMessage(MessageExt replyMsg) {
correlationId, bornHost);
}
}

private RemotingCommand updateMessageRequestMode(ChannelHandlerContext ctx,
RemotingCommand request) {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
final SetMessageRequestModeRequestBody requestBody = SetMessageRequestModeRequestBody.decode(request.getBody(), SetMessageRequestModeRequestBody.class);

final String group = requestBody.getConsumerGroup();
final String topic = requestBody.getTopic();
final MessageRequestMode requestMode = requestBody.getMode();
final int popShareQueueNum = requestBody.getPopShareQueueNum();
this.mqClientFactory.updateRebalanceByBrokerAndClientMap(group, topic, requestMode, popShareQueueNum);

response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
return response;
}
}
16 changes: 16 additions & 0 deletions client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,7 @@
import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody;
import org.apache.rocketmq.remoting.protocol.body.LockBatchResponseBody;
import org.apache.rocketmq.remoting.protocol.body.MessageRequestModeSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
import org.apache.rocketmq.remoting.protocol.body.QueryAssignmentRequestBody;
Expand Down Expand Up @@ -3097,6 +3098,21 @@ public void setMessageRequestMode(final String brokerAddr, final String topic, f
}
}

public MessageRequestModeSerializeWrapper getAllMessageRequestMode(final String brokerAddr, final long timeoutMillis) throws MQBrokerException,
RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_MESSAGE_REQUEST_MODE, null);
RemotingCommand response = this.remotingClient.invokeSync(brokerAddr, request, timeoutMillis);
assert response != null;
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
return MessageRequestModeSerializeWrapper.decode(response.getBody(), MessageRequestModeSerializeWrapper.class);
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr);
}

public TopicConfigAndQueueMapping getTopicConfig(final String brokerAddr, String topic,
long timeoutMillis) throws InterruptedException,
RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException {
Expand Down
Loading
Loading