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 2018/07/16 08:55:03 UTC
[GitHub] vongosling closed pull request #203: [ROCKETMQ-331] Make DLQ queue
nums configurable
vongosling closed pull request #203: [ROCKETMQ-331] Make DLQ queue nums configurable
URL: https://github.com/apache/rocketmq/pull/203
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
index aa072e8b6..c4edb3e1d 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AbstractSendMessageProcessor.java
@@ -55,7 +55,6 @@
public abstract class AbstractSendMessageProcessor implements NettyRequestProcessor {
protected static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
- protected final static int DLQ_NUMS_PER_GROUP = 1;
protected final BrokerController brokerController;
protected final Random random = new Random(System.currentTimeMillis());
protected final SocketAddress storeHost;
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
index b7e7a6187..58ce359da 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
@@ -181,11 +181,11 @@ private RemotingCommand consumerSendMsgBack(final ChannelHandlerContext ctx, fin
if (msgExt.getReconsumeTimes() >= maxReconsumeTimes
|| delayLevel < 0) {
newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
- queueIdInt = Math.abs(this.random.nextInt() % 99999999) % DLQ_NUMS_PER_GROUP;
+ queueIdInt = Math.abs(this.random.nextInt() % 99999999) % this.brokerController.getBrokerConfig().getDlqNumsPerGroup();
topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic,
- DLQ_NUMS_PER_GROUP,
- PermName.PERM_WRITE, 0
+ this.brokerController.getBrokerConfig().getDlqNumsPerGroup(),
+ this.brokerController.getBrokerConfig().getDlqPerm(), 0
);
if (null == topicConfig) {
response.setCode(ResponseCode.SYSTEM_ERROR);
@@ -270,10 +270,10 @@ private boolean handleRetryAndDLQ(SendMessageRequestHeader requestHeader, Remoti
int reconsumeTimes = requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes();
if (reconsumeTimes >= maxReconsumeTimes) {
newTopic = MixAll.getDLQTopic(groupName);
- int queueIdInt = Math.abs(this.random.nextInt() % 99999999) % DLQ_NUMS_PER_GROUP;
+ int queueIdInt = Math.abs(this.random.nextInt() % 99999999) % this.brokerController.getBrokerConfig().getDlqNumsPerGroup();
topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic,
- DLQ_NUMS_PER_GROUP,
- PermName.PERM_WRITE, 0
+ this.brokerController.getBrokerConfig().getDlqNumsPerGroup(),
+ this.brokerController.getBrokerConfig().getDlqPerm(), 0
);
msg.setTopic(newTopic);
msg.setQueueId(queueIdInt);
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
index 56abf084a..fc8a3c85e 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
@@ -17,16 +17,28 @@
package org.apache.rocketmq.broker;
+import org.apache.rocketmq.broker.processor.SendMessageProcessor;
import java.io.File;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.constant.PermName;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.store.AppendMessageResult;
+import org.apache.rocketmq.store.AppendMessageStatus;
+import org.apache.rocketmq.store.MessageExtBrokerInner;
+import org.apache.rocketmq.store.PutMessageResult;
+import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
public class BrokerControllerTest {
@@ -42,6 +54,20 @@ public void testBrokerRestart() throws Exception {
brokerController.shutdown();
}
+ @Test
+ public void testBrokerConfigDLQ() throws Exception {
+ BrokerConfig brokerConfig = new BrokerConfig();
+ brokerConfig.setDlqPerm(PermName.PERM_READ | PermName.PERM_WRITE);
+ brokerConfig.setDlqNumsPerGroup(8);
+ BrokerController brokerController = new BrokerController(
+ brokerConfig,
+ new NettyServerConfig(),
+ new NettyClientConfig(),
+ new MessageStoreConfig());
+ assertThat(brokerController.getBrokerConfig().getDlqPerm() == (PermName.PERM_WRITE | PermName.PERM_READ));
+ assertThat(brokerController.getBrokerConfig().getDlqNumsPerGroup() == 8);
+ }
+
@After
public void destroy() {
UtilAll.deleteFile(new File(new MessageStoreConfig().getStorePathRootDir()));
diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
index e4486da0d..99148f7d2 100644
--- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -132,6 +132,10 @@
private boolean filterSupportRetry = false;
private boolean enablePropertyFilter = false;
+ private int dlqNumsPerGroup = 1;
+
+ private int dlqPerm = PermName.PERM_WRITE;
+
private boolean compressedRegister = false;
private boolean forceRegister = true;
@@ -627,6 +631,22 @@ public void setEnablePropertyFilter(boolean enablePropertyFilter) {
this.enablePropertyFilter = enablePropertyFilter;
}
+ public int getDlqNumsPerGroup() {
+ return dlqNumsPerGroup;
+ }
+
+ public void setDlqNumsPerGroup(int dlqNumsPerGroup) {
+ this.dlqNumsPerGroup = dlqNumsPerGroup;
+ }
+
+ public int getDlqPerm() {
+ return dlqPerm;
+ }
+
+ public void setDlqPerm(int dlqPerm) {
+ this.dlqPerm = dlqPerm;
+ }
+
public boolean isCompressedRegister() {
return compressedRegister;
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
With regards,
Apache Git Services