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

[ISSUE #5066] Fix only individual queues are consumed in high tps scenarios (#5067)

Browse Source
This commit is contained in:
guyinyou
2022-09-14 14:47:20 +08:00
committed by GitHub
parent 5a881bad39
commit 6efbe68241
3 changed files with 6 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/DefaultPullMessageResultHandler.java
+1 -1
View File
@@ -172,7 +172,7 @@ public class DefaultPullMessageResultHandler implements PullMessageResultHandler
}
private boolean channelIsWritable(Channel channel, PullMessageRequestHeader requestHeader) {
if (this.brokerController.getBrokerConfig().isNetWorkFlowController()) {
if (this.brokerController.getBrokerConfig().isEnableNetWorkFlowControl()) {
if (!channel.isWritable()) {
log.warn("channel {} not writable ,cid {}", channel.remoteAddress(), requestHeader.getConsumerGroup());
return false;
broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java
-2
View File
@@ -83,10 +83,8 @@ public class PullMessageProcessorTest {
SubscriptionGroupManager subscriptionGroupManager = new SubscriptionGroupManager(brokerController);
pullMessageProcessor = new PullMessageProcessor(brokerController);
Channel mockChannel = mock(Channel.class);
when(mockChannel.isWritable()).thenReturn(true);
when(mockChannel.remoteAddress()).thenReturn(new InetSocketAddress(1024));
when(handlerContext.channel()).thenReturn(mockChannel);
when(handlerContext.channel().isWritable()).thenReturn(true);
when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager);
brokerController.getTopicConfigManager().getTopicConfigTable().put(topic, new TopicConfig());
clientChannelInfo = new ClientChannelInfo(mockChannel);
common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+5 -5
View File
@@ -191,7 +191,7 @@ public class BrokerConfig extends BrokerIdentity {
*/
private long brokerNotActiveTimeoutMillis = 10 * 1000;
private boolean netWorkFlowController = true;
private boolean enableNetWorkFlowControl = false;
private int popPollingSize = 1024;
private int popPollingMapSize = 100000;
@@ -1156,12 +1156,12 @@ public class BrokerConfig extends BrokerIdentity {
this.brokerNotActiveTimeoutMillis = brokerNotActiveTimeoutMillis;
}
public boolean isNetWorkFlowController() {
return netWorkFlowController;
public boolean isEnableNetWorkFlowControl() {
return enableNetWorkFlowControl;
}
public void setNetWorkFlowController(boolean netWorkFlowController) {
this.netWorkFlowController = netWorkFlowController;
public void setEnableNetWorkFlowControl(boolean enableNetWorkFlowControl) {
this.enableNetWorkFlowControl = enableNetWorkFlowControl;
}
public boolean isRealTimeNotifyConsumerChange() {