opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 05:46:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #7351] Allow mqadmin to operate slave nodes

Browse Source 
Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>
This commit is contained in:
guyinyou
2023-09-13 16:08:03 +08:00
committed by GitHub
parent fd32dae2ab
commit 4a8e0d5b85
2 changed files with 0 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
-12
View File
@@ -406,9 +406,6 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
if (validateSlave(response)) {
return response;
}
final CreateTopicRequestHeader requestHeader =
(CreateTopicRequestHeader) request.decodeCommandCustomHeader(CreateTopicRequestHeader.class);
@@ -519,9 +516,6 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
private synchronized RemotingCommand deleteTopic(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
if (validateSlave(response)) {
return response;
}
DeleteTopicRequestHeader requestHeader =
(DeleteTopicRequestHeader) request.decodeCommandCustomHeader(DeleteTopicRequestHeader.class);
@@ -1413,9 +1407,6 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
private RemotingCommand updateAndCreateSubscriptionGroup(ChannelHandlerContext ctx, RemotingCommand request)
throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
if (validateSlave(response)) {
return response;
}
LOGGER.info("AdminBrokerProcessor#updateAndCreateSubscriptionGroup called by {}",
RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
@@ -1480,9 +1471,6 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
private RemotingCommand deleteSubscriptionGroup(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
if (validateSlave(response)) {
return response;
}
DeleteSubscriptionGroupRequestHeader requestHeader =
(DeleteSubscriptionGroupRequestHeader) request.decodeCommandCustomHeader(DeleteSubscriptionGroupRequestHeader.class);
broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
-106
View File
@@ -76,7 +76,6 @@ import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfi
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.logfile.DefaultMappedFile;
import org.apache.rocketmq.store.stats.BrokerStats;
@@ -250,32 +249,6 @@ public class AdminBrokerProcessorTest {
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testUpdateAndCreateTopicOnSlaveInRocksdb() throws Exception {
if (notToBeExecuted()) {
return;
}
initRocksdbTopicManager();
testUpdateAndCreateTopicOnSlave();
}
@Test
public void testUpdateAndCreateTopicOnSlave() throws Exception {
// setup
MessageStoreConfig messageStoreConfig = mock(MessageStoreConfig.class);
when(messageStoreConfig.getBrokerRole()).thenReturn(BrokerRole.SLAVE);
defaultMessageStore = mock(DefaultMessageStore.class);
when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
// test on slave
String topic = "TEST_CREATE_TOPIC";
RemotingCommand request = buildCreateTopicRequest(topic);
RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
assertThat(response.getRemark()).isEqualTo("Can't modify topic or subscription group from slave broker, " +
"please execute it from master broker.");
}
@Test
public void testDeleteTopicInRocksdb() throws Exception {
if (notToBeExecuted()) {
@@ -301,31 +274,6 @@ public class AdminBrokerProcessorTest {
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testDeleteTopicOnSlaveInRocksdb() throws Exception {
if (notToBeExecuted()) {
return;
}
initRocksdbTopicManager();
testDeleteTopicOnSlave();
}
@Test
public void testDeleteTopicOnSlave() throws Exception {
// setup
MessageStoreConfig messageStoreConfig = mock(MessageStoreConfig.class);
when(messageStoreConfig.getBrokerRole()).thenReturn(BrokerRole.SLAVE);
defaultMessageStore = mock(DefaultMessageStore.class);
when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
String topic = "TEST_DELETE_TOPIC";
RemotingCommand request = buildDeleteTopicRequest(topic);
RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
assertThat(response.getRemark()).isEqualTo("Can't modify topic or subscription group from slave broker, " +
"please execute it from master broker.");
}
@Test
public void testDeleteWithPopRetryTopic() throws Exception {
String topic = "topicA";
@@ -538,36 +486,6 @@ public class AdminBrokerProcessorTest {
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testUpdateAndCreateSubscriptionGroupOnSlaveInRocksdb() throws Exception {
initRocksdbSubscriptionManager();
testUpdateAndCreateSubscriptionGroupOnSlave();
}
@Test
public void testUpdateAndCreateSubscriptionGroupOnSlave() throws RemotingCommandException {
// Setup
MessageStoreConfig messageStoreConfig = mock(MessageStoreConfig.class);
when(messageStoreConfig.getBrokerRole()).thenReturn(BrokerRole.SLAVE);
defaultMessageStore = mock(DefaultMessageStore.class);
when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
// Test
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP, null);
SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
subscriptionGroupConfig.setBrokerId(1);
subscriptionGroupConfig.setGroupName("groupId");
subscriptionGroupConfig.setConsumeEnable(Boolean.TRUE);
subscriptionGroupConfig.setConsumeBroadcastEnable(Boolean.TRUE);
subscriptionGroupConfig.setRetryMaxTimes(111);
subscriptionGroupConfig.setConsumeFromMinEnable(Boolean.TRUE);
request.setBody(JSON.toJSON(subscriptionGroupConfig).toString().getBytes());
RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
assertThat(response.getRemark()).isEqualTo("Can't modify topic or subscription group from slave broker, " +
"please execute it from master broker.");
}
@Test
public void testGetAllSubscriptionGroupInRocksdb() throws Exception {
initRocksdbSubscriptionManager();
@@ -596,30 +514,6 @@ public class AdminBrokerProcessorTest {
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testDeleteSubscriptionGroupOnSlaveInRocksdb() throws Exception {
initRocksdbSubscriptionManager();
testDeleteSubscriptionGroupOnSlave();
}
@Test
public void testDeleteSubscriptionGroupOnSlave() throws RemotingCommandException {
// Setup
MessageStoreConfig messageStoreConfig = mock(MessageStoreConfig.class);
when(messageStoreConfig.getBrokerRole()).thenReturn(BrokerRole.SLAVE);
defaultMessageStore = mock(DefaultMessageStore.class);
when(brokerController.getMessageStoreConfig()).thenReturn(messageStoreConfig);
// Test
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTIONGROUP, null);
request.addExtField("groupName", "GID-Group-Name");
request.addExtField("removeOffset", "true");
RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
assertThat(response.getRemark()).isEqualTo("Can't modify topic or subscription group from slave broker, " +
"please execute it from master broker.");
}
@Test
public void testGetTopicStatsInfo() throws RemotingCommandException {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_TOPIC_STATS_INFO, null);