iterator = Splitter.on(File.separatorChar).split(path).iterator();
+ boolean found = false;
+ File dir = dstDir;
+ while (iterator.hasNext()) {
+ String current = iterator.next();
+ if (!found && flag.equals(current)) {
+ found = true;
+ if (into) {
+ dir = new File(dir, flag);
+ if (!dir.exists()) {
+ Assert.assertTrue(dir.mkdirs());
+ }
+ }
+ continue;
+ }
+
+ if (found) {
+ if (!iterator.hasNext()) {
+ dir = new File(dir, current);
+ } else {
+ dir = new File(dir, current);
+ if (!dir.exists()) {
+ Assert.assertTrue(dir.mkdir());
+ }
+ }
+ }
+ }
+
+ Assert.assertTrue(dir.createNewFile());
+ byte[] buffer = new byte[4096];
+ BufferedInputStream bis = new BufferedInputStream(src);
+ int len = 0;
+ try (BufferedOutputStream bos = new BufferedOutputStream(Files.newOutputStream(dir.toPath()))) {
+ while ((len = bis.read(buffer)) > 0) {
+ bos.write(buffer, 0, len);
+ }
+ }
+ }
+
+ public static void recursiveDelete(File file) {
+ if (file.isFile()) {
+ file.delete();
+ } else {
+ File[] files = file.listFiles();
+ if (null != files) {
+ for (File f : files) {
+ recursiveDelete(f);
+ }
+ }
+ file.delete();
+ }
+ }
+
+ public static File copyResources(String folder) throws IOException {
+ return copyResources(folder, false);
+ }
+
+ public static File copyResources(String folder, boolean into) throws IOException {
+ File home = new File(System.getProperty("java.io.tmpdir"), UUID.randomUUID().toString().replace('-', '_'));
+ if (!home.exists()) {
+ Assert.assertTrue(home.mkdirs());
+ }
+ PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver(AclTestHelper.class.getClassLoader());
+ Resource[] resources = resolver.getResources(String.format("classpath:%s/**/*", folder));
+ for (Resource resource : resources) {
+ if (!resource.isReadable()) {
+ continue;
+ }
+ String description = resource.getDescription();
+ int start = description.indexOf('[');
+ int end = description.lastIndexOf(']');
+ String path = description.substring(start + 1, end);
+ try (InputStream inputStream = resource.getInputStream()) {
+ copyTo(path, inputStream, home, folder, into);
+ }
+ }
+ return home;
+ }
+}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessControlFlowTest.java b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessControlFlowTest.java
new file mode 100644
index 00000000000..5193457146d
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessControlFlowTest.java
@@ -0,0 +1,311 @@
+/*
+ * 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.acl.plain;
+
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.AclConstants;
+import org.apache.rocketmq.acl.common.AclException;
+import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.common.AclConfig;
+import org.apache.rocketmq.common.PlainAccessConfig;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeaderV2;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * In this class, we'll test the following scenarios, each containing several consecutive operations on ACL,
+ *
like updating and deleting ACL, changing config files and checking validations.
+ *
Case 1: Only conf/plain_acl.yml exists;
+ *
Case 2: Only conf/acl/plain_acl.yml exists;
+ *
Case 3: Both conf/plain_acl.yml and conf/acl/plain_acl.yml exists.
+ */
+public class PlainAccessControlFlowTest {
+ public static final String DEFAULT_TOPIC = "topic-acl";
+
+ public static final String DEFAULT_GROUP = "GID_acl";
+
+ public static final String DEFAULT_PRODUCER_AK = "ak11111";
+ public static final String DEFAULT_PRODUCER_SK = "1234567";
+
+ public static final String DEFAULT_CONSUMER_SK = "7654321";
+ public static final String DEFAULT_CONSUMER_AK = "ak22222";
+
+ public static final String DEFAULT_GLOBAL_WHITE_ADDR = "172.16.123.123";
+ public static final List DEFAULT_GLOBAL_WHITE_ADDRS_LIST = Collections.singletonList(DEFAULT_GLOBAL_WHITE_ADDR);
+
+ @Test
+ public void testEmptyAclFolderCase() throws NoSuchFieldException, IllegalAccessException,
+ IOException {
+ String folder = "empty_acl_folder_conf";
+ File home = AclTestHelper.copyResources(folder);
+ System.setProperty("rocketmq.home.dir", home.getAbsolutePath());
+ PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+ checkDefaultAclFileExists();
+ testValidationAfterConsecutiveUpdates(plainAccessValidator);
+ testValidationAfterConfigFileChanged(plainAccessValidator);
+ AclTestHelper.recursiveDelete(home);
+ }
+
+ @Test
+ public void testOnlyAclFolderCase() throws NoSuchFieldException, IllegalAccessException, IOException {
+ String folder = "only_acl_folder_conf";
+ File home = AclTestHelper.copyResources(folder);
+ System.setProperty("rocketmq.home.dir", home.getAbsolutePath());
+ PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+ checkDefaultAclFileExists();
+ testValidationAfterConsecutiveUpdates(plainAccessValidator);
+ testValidationAfterConfigFileChanged(plainAccessValidator);
+ AclTestHelper.recursiveDelete(home);
+ }
+
+ @Test
+ public void testBothAclFileAndFolderCase() throws NoSuchFieldException, IllegalAccessException,
+ IOException {
+ String folder = "both_acl_file_folder_conf";
+ File root = AclTestHelper.copyResources(folder);
+ System.setProperty("rocketmq.home.dir", root.getAbsolutePath());
+ PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+ checkDefaultAclFileExists();
+ testValidationAfterConsecutiveUpdates(plainAccessValidator);
+ testValidationAfterConfigFileChanged(plainAccessValidator);
+ AclTestHelper.recursiveDelete(root);
+ }
+
+ private void testValidationAfterConfigFileChanged(
+ PlainAccessValidator plainAccessValidator) throws NoSuchFieldException, IllegalAccessException {
+ PlainAccessConfig producerAccessConfig = generateProducerAccessConfig();
+ PlainAccessConfig consumerAccessConfig = generateConsumerAccessConfig();
+ List plainAccessConfigList = new LinkedList<>();
+ plainAccessConfigList.add(producerAccessConfig);
+ plainAccessConfigList.add(consumerAccessConfig);
+ PlainAccessData ymlMap = new PlainAccessData();
+ ymlMap.setAccounts(plainAccessConfigList);
+
+ // write prepared PlainAccessConfigs to file
+ final String aclConfigFile = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
+ AclUtils.writeDataObject(aclConfigFile, ymlMap);
+
+ loadConfigFile(plainAccessValidator, aclConfigFile);
+
+ // check if added successfully
+ final AclConfig allAclConfig = plainAccessValidator.getAllAclConfig();
+ final List plainAccessConfigs = allAclConfig.getPlainAccessConfigs();
+ checkPlainAccessConfig(producerAccessConfig, plainAccessConfigs);
+ checkPlainAccessConfig(consumerAccessConfig, plainAccessConfigs);
+
+ //delete consumer account
+ plainAccessConfigList.remove(consumerAccessConfig);
+ AclUtils.writeDataObject(aclConfigFile, ymlMap);
+
+ loadConfigFile(plainAccessValidator, aclConfigFile);
+
+ // sending messages will be successful using prepared credentials
+ SessionCredentials producerCredential = new SessionCredentials(DEFAULT_PRODUCER_AK, DEFAULT_PRODUCER_SK);
+ AclClientRPCHook producerHook = new AclClientRPCHook(producerCredential);
+ validateSendMessage(RequestCode.SEND_MESSAGE, DEFAULT_TOPIC, producerHook, "", plainAccessValidator);
+ validateSendMessage(RequestCode.SEND_MESSAGE_V2, DEFAULT_TOPIC, producerHook, "", plainAccessValidator);
+
+ // consuming messages will be failed for account has been deleted
+ SessionCredentials consumerCredential = new SessionCredentials(DEFAULT_CONSUMER_AK, DEFAULT_CONSUMER_SK);
+ AclClientRPCHook consumerHook = new AclClientRPCHook(consumerCredential);
+ boolean isConsumeFailed = false;
+ try {
+ validatePullMessage(DEFAULT_TOPIC, DEFAULT_GROUP, consumerHook, "", plainAccessValidator);
+ } catch (AclException e) {
+ isConsumeFailed = true;
+ }
+ Assert.assertTrue("Message should not be consumed after account deleted", isConsumeFailed);
+
+ }
+
+ private void testValidationAfterConsecutiveUpdates(
+ PlainAccessValidator plainAccessValidator) throws NoSuchFieldException, IllegalAccessException {
+ PlainAccessConfig producerAccessConfig = generateProducerAccessConfig();
+ plainAccessValidator.updateAccessConfig(producerAccessConfig);
+
+ PlainAccessConfig consumerAccessConfig = generateConsumerAccessConfig();
+ plainAccessValidator.updateAccessConfig(consumerAccessConfig);
+
+ plainAccessValidator.updateGlobalWhiteAddrsConfig(DEFAULT_GLOBAL_WHITE_ADDRS_LIST, null);
+
+ // check if the above config updated successfully
+ final AclConfig allAclConfig = plainAccessValidator.getAllAclConfig();
+ final List plainAccessConfigs = allAclConfig.getPlainAccessConfigs();
+ checkPlainAccessConfig(producerAccessConfig, plainAccessConfigs);
+ checkPlainAccessConfig(consumerAccessConfig, plainAccessConfigs);
+
+ Assert.assertEquals(DEFAULT_GLOBAL_WHITE_ADDRS_LIST, allAclConfig.getGlobalWhiteAddrs());
+
+ // check sending and consuming messages
+ SessionCredentials producerCredential = new SessionCredentials(DEFAULT_PRODUCER_AK, DEFAULT_PRODUCER_SK);
+ AclClientRPCHook producerHook = new AclClientRPCHook(producerCredential);
+ validateSendMessage(RequestCode.SEND_MESSAGE, DEFAULT_TOPIC, producerHook, "", plainAccessValidator);
+ validateSendMessage(RequestCode.SEND_MESSAGE_V2, DEFAULT_TOPIC, producerHook, "", plainAccessValidator);
+
+ SessionCredentials consumerCredential = new SessionCredentials(DEFAULT_CONSUMER_AK, DEFAULT_CONSUMER_SK);
+ AclClientRPCHook consumerHook = new AclClientRPCHook(consumerCredential);
+ validatePullMessage(DEFAULT_TOPIC, DEFAULT_GROUP, consumerHook, "", plainAccessValidator);
+
+ // load from file
+ loadConfigFile(plainAccessValidator,
+ System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml");
+ SessionCredentials unmatchedCredential = new SessionCredentials("non_exists_sk", "non_exists_sk");
+ AclClientRPCHook dummyHook = new AclClientRPCHook(unmatchedCredential);
+ validateSendMessage(RequestCode.SEND_MESSAGE, DEFAULT_TOPIC, dummyHook, DEFAULT_GLOBAL_WHITE_ADDR, plainAccessValidator);
+ validateSendMessage(RequestCode.SEND_MESSAGE_V2, DEFAULT_TOPIC, dummyHook, DEFAULT_GLOBAL_WHITE_ADDR, plainAccessValidator);
+ validatePullMessage(DEFAULT_TOPIC, DEFAULT_GROUP, dummyHook, DEFAULT_GLOBAL_WHITE_ADDR, plainAccessValidator);
+
+ //recheck after reloading
+ validateSendMessage(RequestCode.SEND_MESSAGE, DEFAULT_TOPIC, producerHook, "", plainAccessValidator);
+ validateSendMessage(RequestCode.SEND_MESSAGE_V2, DEFAULT_TOPIC, producerHook, "", plainAccessValidator);
+ validatePullMessage(DEFAULT_TOPIC, DEFAULT_GROUP, consumerHook, "", plainAccessValidator);
+
+ }
+
+ private void loadConfigFile(PlainAccessValidator plainAccessValidator,
+ String configFileName) throws NoSuchFieldException, IllegalAccessException {
+ Class clazz = PlainAccessValidator.class;
+ Field f = clazz.getDeclaredField("aclPlugEngine");
+ f.setAccessible(true);
+ PlainPermissionManager aclPlugEngine = (PlainPermissionManager) f.get(plainAccessValidator);
+ aclPlugEngine.load(configFileName);
+ }
+
+ private PlainAccessConfig generateConsumerAccessConfig() {
+ PlainAccessConfig plainAccessConfig2 = new PlainAccessConfig();
+ plainAccessConfig2.setAccessKey(DEFAULT_CONSUMER_AK);
+ plainAccessConfig2.setSecretKey(DEFAULT_CONSUMER_SK);
+ plainAccessConfig2.setAdmin(false);
+ plainAccessConfig2.setDefaultTopicPerm(AclConstants.DENY);
+ plainAccessConfig2.setDefaultGroupPerm(AclConstants.DENY);
+ plainAccessConfig2.setTopicPerms(Collections.singletonList(DEFAULT_TOPIC + "=" + AclConstants.SUB));
+ plainAccessConfig2.setGroupPerms(Collections.singletonList(DEFAULT_GROUP + "=" + AclConstants.SUB));
+ return plainAccessConfig2;
+ }
+
+ private PlainAccessConfig generateProducerAccessConfig() {
+ PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
+ plainAccessConfig.setAccessKey(DEFAULT_PRODUCER_AK);
+ plainAccessConfig.setSecretKey(DEFAULT_PRODUCER_SK);
+ plainAccessConfig.setAdmin(false);
+ plainAccessConfig.setDefaultTopicPerm(AclConstants.DENY);
+ plainAccessConfig.setDefaultGroupPerm(AclConstants.DENY);
+ plainAccessConfig.setTopicPerms(Collections.singletonList(DEFAULT_TOPIC + "=" + AclConstants.PUB));
+ return plainAccessConfig;
+ }
+
+ public void validatePullMessage(String topic,
+ String group,
+ AclClientRPCHook aclClientRPCHook,
+ String remoteAddr,
+ PlainAccessValidator plainAccessValidator) {
+ PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
+ pullMessageRequestHeader.setTopic(topic);
+ pullMessageRequestHeader.setConsumerGroup(group);
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE,
+ pullMessageRequestHeader);
+ aclClientRPCHook.doBeforeRequest(remoteAddr, remotingCommand);
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ try {
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(
+ RemotingCommand.decode(buf), remoteAddr);
+ plainAccessValidator.validate(accessResource);
+ } catch (RemotingCommandException e) {
+ e.printStackTrace();
+ Assert.fail("Should not throw RemotingCommandException");
+ }
+ }
+
+ public void validateSendMessage(int requestCode,
+ String topic,
+ AclClientRPCHook aclClientRPCHook,
+ String remoteAddr,
+ PlainAccessValidator plainAccessValidator) {
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic(topic);
+ RemotingCommand remotingCommand;
+ if (RequestCode.SEND_MESSAGE == requestCode) {
+ remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+ } else {
+ remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2,
+ SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(messageRequestHeader));
+ }
+
+ aclClientRPCHook.doBeforeRequest(remoteAddr, remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ try {
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(
+ RemotingCommand.decode(buf), remoteAddr);
+ plainAccessValidator.validate(accessResource);
+ } catch (RemotingCommandException e) {
+ e.printStackTrace();
+ Assert.fail("Should not throw RemotingCommandException");
+ }
+ }
+
+ private void checkPlainAccessConfig(final PlainAccessConfig plainAccessConfig,
+ final List plainAccessConfigs) {
+ for (PlainAccessConfig config : plainAccessConfigs) {
+ if (config.getAccessKey().equals(plainAccessConfig.getAccessKey())) {
+ Assert.assertEquals(plainAccessConfig.getSecretKey(), config.getSecretKey());
+ Assert.assertEquals(plainAccessConfig.isAdmin(), config.isAdmin());
+ Assert.assertEquals(plainAccessConfig.getDefaultGroupPerm(), config.getDefaultGroupPerm());
+ Assert.assertEquals(plainAccessConfig.getDefaultGroupPerm(), config.getDefaultGroupPerm());
+ Assert.assertEquals(plainAccessConfig.getWhiteRemoteAddress(), config.getWhiteRemoteAddress());
+ if (null != plainAccessConfig.getTopicPerms()) {
+ Assert.assertNotNull(config.getTopicPerms());
+ Assert.assertTrue(config.getTopicPerms().containsAll(plainAccessConfig.getTopicPerms()));
+ }
+ if (null != plainAccessConfig.getGroupPerms()) {
+ Assert.assertNotNull(config.getGroupPerms());
+ Assert.assertTrue(config.getGroupPerms().containsAll(plainAccessConfig.getGroupPerms()));
+ }
+ }
+ }
+ }
+
+ private void checkDefaultAclFileExists() {
+ boolean isExists = Files.exists(Paths.get(System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf" + File.separator + "plain_acl.yml"));
+ Assert.assertTrue("default acl config file should exist", isExists);
+ }
+
+}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java
index 62d98572053..ef0cffbdcc8 100644
--- a/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java
+++ b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java
@@ -16,9 +16,11 @@
*/
package org.apache.rocketmq.acl.plain;
+import com.google.common.base.Joiner;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
@@ -33,24 +35,25 @@
import org.apache.rocketmq.acl.common.AclUtils;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.common.AclConfig;
-import org.apache.rocketmq.common.DataVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.PlainAccessConfig;
-import org.apache.rocketmq.common.protocol.RequestCode;
-import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
-import org.apache.rocketmq.common.protocol.header.GetConsumerListByGroupRequestHeader;
-import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
-import org.apache.rocketmq.common.protocol.header.QueryMessageRequestHeader;
-import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
-import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeaderV2;
-import org.apache.rocketmq.common.protocol.header.UnregisterClientRequestHeader;
-import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
-import org.apache.rocketmq.common.protocol.heartbeat.ConsumerData;
-import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
-import org.apache.rocketmq.common.protocol.heartbeat.ProducerData;
-import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.header.ConsumerSendMsgBackRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeaderV2;
+import org.apache.rocketmq.remoting.protocol.header.UnregisterClientRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.UpdateConsumerOffsetRequestHeader;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData;
+import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -61,11 +64,13 @@ public class PlainAccessValidatorTest {
private AclClientRPCHook aclClient;
private SessionCredentials sessionCredentials;
+ private File confHome;
@Before
- public void init() {
- File file = new File("src/test/resources");
- System.setProperty("rocketmq.home.dir", file.getAbsolutePath());
+ public void init() throws IOException {
+ String folder = "conf";
+ confHome = AclTestHelper.copyResources(folder, true);
+ System.setProperty("rocketmq.home.dir", confHome.getAbsolutePath());
plainAccessValidator = new PlainAccessValidator();
sessionCredentials = new SessionCredentials();
sessionCredentials.setAccessKey("RocketMQ");
@@ -74,6 +79,11 @@ public void init() {
aclClient = new AclClientRPCHook(sessionCredentials);
}
+ @After
+ public void cleanUp() {
+ AclTestHelper.recursiveDelete(confHome);
+ }
+
@Test
public void contentTest() {
SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
@@ -110,7 +120,7 @@ public void validateTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -132,7 +142,28 @@ public void validateSendMessageTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
+ plainAccessValidator.validate(accessResource);
+ } catch (RemotingCommandException e) {
+ e.printStackTrace();
+
+ Assert.fail("Should not throw IOException");
+ }
+ }
+
+ @Test
+ public void validateSendMessageToRetryTopicTest() {
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic(MixAll.getRetryTopic("groupB"));
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ try {
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -144,7 +175,7 @@ public void validateSendMessageTest() {
@Test
public void validateSendMessageV2Test() {
SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
- messageRequestHeader.setTopic("topicC");
+ messageRequestHeader.setTopic("topicB");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(messageRequestHeader));
aclClient.doBeforeRequest("", remotingCommand);
@@ -153,7 +184,28 @@ public void validateSendMessageV2Test() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
+ plainAccessValidator.validate(accessResource);
+ } catch (RemotingCommandException e) {
+ e.printStackTrace();
+
+ Assert.fail("Should not throw IOException");
+ }
+ }
+
+ @Test
+ public void validateSendMessageV2ToRetryTopicTest() {
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic(MixAll.getRetryTopic("groupC"));
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(messageRequestHeader));
+ aclClient.doBeforeRequest("", remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ try {
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6:9876");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -182,7 +234,7 @@ public void validateForAdminCommandWithOutAclRPCHook() {
public void validatePullMessageTest() {
PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
pullMessageRequestHeader.setTopic("topicC");
- pullMessageRequestHeader.setConsumerGroup("consumerGroupA");
+ pullMessageRequestHeader.setConsumerGroup("groupC");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
@@ -190,7 +242,7 @@ public void validatePullMessageTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -203,7 +255,7 @@ public void validatePullMessageTest() {
public void validateConsumeMessageBackTest() {
ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader = new ConsumerSendMsgBackRequestHeader();
consumerSendMsgBackRequestHeader.setOriginTopic("topicC");
- consumerSendMsgBackRequestHeader.setGroup("consumerGroupA");
+ consumerSendMsgBackRequestHeader.setGroup("groupC");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, consumerSendMsgBackRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
@@ -211,7 +263,7 @@ public void validateConsumeMessageBackTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -231,7 +283,7 @@ public void validateQueryMessageTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -268,9 +320,9 @@ public void validateHeartBeatTest() {
Set consumerDataSet = new HashSet<>();
Set subscriptionDataSet = new HashSet<>();
ProducerData producerData = new ProducerData();
- producerData.setGroupName("producerGroupA");
+ producerData.setGroupName("groupB");
ConsumerData consumerData = new ConsumerData();
- consumerData.setGroupName("consumerGroupA");
+ consumerData.setGroupName("groupC");
SubscriptionData subscriptionData = new SubscriptionData();
subscriptionData.setTopic("topicC");
producerDataSet.add(producerData);
@@ -287,7 +339,7 @@ public void validateHeartBeatTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -299,7 +351,7 @@ public void validateHeartBeatTest() {
@Test
public void validateUnRegisterClientTest() {
UnregisterClientRequestHeader unregisterClientRequestHeader = new UnregisterClientRequestHeader();
- unregisterClientRequestHeader.setConsumerGroup("consumerGroupA");
+ unregisterClientRequestHeader.setConsumerGroup("groupB");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT, unregisterClientRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
@@ -307,7 +359,7 @@ public void validateUnRegisterClientTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -319,7 +371,7 @@ public void validateUnRegisterClientTest() {
@Test
public void validateGetConsumerListByGroupTest() {
GetConsumerListByGroupRequestHeader getConsumerListByGroupRequestHeader = new GetConsumerListByGroupRequestHeader();
- getConsumerListByGroupRequestHeader.setConsumerGroup("consumerGroupA");
+ getConsumerListByGroupRequestHeader.setConsumerGroup("groupB");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_LIST_BY_GROUP, getConsumerListByGroupRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
@@ -327,7 +379,7 @@ public void validateGetConsumerListByGroupTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -339,7 +391,7 @@ public void validateGetConsumerListByGroupTest() {
@Test
public void validateUpdateConsumerOffSetTest() {
UpdateConsumerOffsetRequestHeader updateConsumerOffsetRequestHeader = new UpdateConsumerOffsetRequestHeader();
- updateConsumerOffsetRequestHeader.setConsumerGroup("consumerGroupA");
+ updateConsumerOffsetRequestHeader.setConsumerGroup("groupB");
RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.UPDATE_CONSUMER_OFFSET, updateConsumerOffsetRequestHeader);
aclClient.doBeforeRequest("", remotingCommand);
ByteBuffer buf = remotingCommand.encodeHeader();
@@ -347,7 +399,7 @@ public void validateUpdateConsumerOffSetTest() {
buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
buf.position(0);
try {
- PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
plainAccessValidator.validate(accessResource);
} catch (RemotingCommandException e) {
e.printStackTrace();
@@ -427,8 +479,12 @@ public void validateGetAllTopicConfigTest() {
@Test
public void addAccessAclYamlConfigTest() throws InterruptedException {
- String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+ String backupFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl_bak.yml".replace("/", File.separator);
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(backupFileName, PlainAccessData.class);
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
plainAccessConfig.setAccessKey("rocketmq3");
@@ -436,11 +492,11 @@ public void addAccessAclYamlConfigTest() throws InterruptedException {
plainAccessConfig.setWhiteRemoteAddress("192.168.0.*");
plainAccessConfig.setDefaultGroupPerm("PUB");
plainAccessConfig.setDefaultTopicPerm("SUB");
- List topicPerms = new ArrayList();
+ List topicPerms = new ArrayList<>();
topicPerms.add("topicC=PUB|SUB");
topicPerms.add("topicB=PUB");
plainAccessConfig.setTopicPerms(topicPerms);
- List groupPerms = new ArrayList();
+ List groupPerms = new ArrayList<>();
groupPerms.add("groupB=PUB|SUB");
groupPerms.add("groupC=DENY");
plainAccessConfig.setGroupPerms(groupPerms);
@@ -473,9 +529,9 @@ public void addAccessAclYamlConfigTest() throws InterruptedException {
Assert.assertEquals(((List) verifyMap.get(AclConstants.CONFIG_GROUP_PERMS)).size(), 2);
String aclFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map readableMap = AclUtils.getYamlDataObject(aclFileName, Map.class);
- List> dataVersions = (List>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
- Assert.assertEquals(1, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
+ PlainAccessData readableMap = AclUtils.getYamlDataObject(aclFileName, PlainAccessData.class);
+ List dataVersions = readableMap.getDataVersion();
+ Assert.assertEquals(1L, dataVersions.get(0).getCounter());
AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
}
@@ -499,16 +555,21 @@ public void getAccessAclYamlConfigTest() {
Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_ADMIN_ROLE), true);
Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_WHITE_ADDR), "192.168.1.*");
- String aclFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl.yml";
+ String aclFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
Map dataVersionMap = plainAccessValidator.getAllAclConfigVersion();
DataVersion dataVersion = dataVersionMap.get(aclFileName);
Assert.assertEquals(0, dataVersion.getCounter().get());
}
@Test
- public void updateAccessAclYamlConfigTest() throws InterruptedException{
- String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+ public void updateAccessAclYamlConfigTest() throws InterruptedException {
+ String backupFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl_bak.yml".replace("/", File.separator);
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(backupFileName, PlainAccessData.class);
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
plainAccessConfig.setAccessKey("rocketmq3");
@@ -516,11 +577,11 @@ public void updateAccessAclYamlConfigTest() throws InterruptedException{
plainAccessConfig.setWhiteRemoteAddress("192.168.0.*");
plainAccessConfig.setDefaultGroupPerm("PUB");
plainAccessConfig.setDefaultTopicPerm("SUB");
- List topicPerms = new ArrayList();
+ List topicPerms = new ArrayList<>();
topicPerms.add("topicC=PUB|SUB");
topicPerms.add("topicB=PUB");
plainAccessConfig.setTopicPerms(topicPerms);
- List groupPerms = new ArrayList();
+ List groupPerms = new ArrayList<>();
groupPerms.add("groupB=PUB|SUB");
groupPerms.add("groupC=DENY");
plainAccessConfig.setGroupPerms(groupPerms);
@@ -536,11 +597,11 @@ public void updateAccessAclYamlConfigTest() throws InterruptedException{
plainAccessConfig1.setWhiteRemoteAddress("192.168.0.*");
plainAccessConfig1.setDefaultGroupPerm("PUB");
plainAccessConfig1.setDefaultTopicPerm("SUB");
- List topicPerms1 = new ArrayList();
+ List topicPerms1 = new ArrayList<>();
topicPerms1.add("topicC=PUB|SUB");
topicPerms1.add("topicB=PUB");
plainAccessConfig1.setTopicPerms(topicPerms1);
- List groupPerms1 = new ArrayList();
+ List groupPerms1 = new ArrayList<>();
groupPerms1.add("groupB=PUB|SUB");
groupPerms1.add("groupC=DENY");
plainAccessConfig1.setGroupPerms(groupPerms1);
@@ -572,18 +633,23 @@ public void updateAccessAclYamlConfigTest() throws InterruptedException{
Assert.assertEquals(((List) verifyMap.get(AclConstants.CONFIG_TOPIC_PERMS)).size(), 2);
Assert.assertEquals(((List) verifyMap.get(AclConstants.CONFIG_GROUP_PERMS)).size(), 2);
- String aclFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map readableMap = AclUtils.getYamlDataObject(aclFileName, Map.class);
- List> dataVersions = (List>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
- Assert.assertEquals(2, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
+ String aclFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData readableMap = AclUtils.getYamlDataObject(aclFileName, PlainAccessData.class);
+ List dataVersions = readableMap.getDataVersion();
+ Assert.assertEquals(2L, dataVersions.get(0).getCounter());
AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
}
@Test
public void deleteAccessAclYamlConfigTest() throws InterruptedException {
- String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+ String backupFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl_bak.yml".replace("/", File.separator);
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(backupFileName, PlainAccessData.class);
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
plainAccessConfig.setAccessKey("rocketmq3");
@@ -591,11 +657,11 @@ public void deleteAccessAclYamlConfigTest() throws InterruptedException {
plainAccessConfig.setWhiteRemoteAddress("192.168.0.*");
plainAccessConfig.setDefaultGroupPerm("PUB");
plainAccessConfig.setDefaultTopicPerm("SUB");
- List topicPerms = new ArrayList();
+ List topicPerms = new ArrayList<>();
topicPerms.add("topicC=PUB|SUB");
topicPerms.add("topicB=PUB");
plainAccessConfig.setTopicPerms(topicPerms);
- List groupPerms = new ArrayList();
+ List groupPerms = new ArrayList<>();
groupPerms.add("groupB=PUB|SUB");
groupPerms.add("groupC=DENY");
plainAccessConfig.setGroupPerms(groupPerms);
@@ -629,25 +695,31 @@ public void deleteAccessAclYamlConfigTest() throws InterruptedException {
@Test
public void updateGlobalWhiteRemoteAddressesTest() throws InterruptedException {
- String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+ String backupFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl_bak.yml".replace("/", File.separator);
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(backupFileName, PlainAccessData.class);
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
List globalWhiteAddrsList = new ArrayList<>();
globalWhiteAddrsList.add("192.168.1.*");
PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
- Assert.assertEquals(plainAccessValidator.updateGlobalWhiteAddrsConfig(globalWhiteAddrsList), true);
+ Assert.assertEquals(plainAccessValidator.updateGlobalWhiteAddrsConfig(globalWhiteAddrsList, null), true);
- String aclFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map readableMap = AclUtils.getYamlDataObject(aclFileName, Map.class);
- List> dataVersions = (List>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
- Assert.assertEquals(1, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
+ String aclFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData readableMap = AclUtils.getYamlDataObject(aclFileName, PlainAccessData.class);
+ List dataVersions = readableMap.getDataVersion();
+ Assert.assertEquals(1L, dataVersions.get(0).getCounter());
AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
}
@Test
public void addYamlConfigTest() throws IOException, InterruptedException {
- String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl_test.yml";
+ String fileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/acl/plain_acl_test.yml".replace("/", File.separator);
File transport = new File(fileName);
transport.delete();
transport.createNewFile();
@@ -687,7 +759,8 @@ public void addYamlConfigTest() throws IOException, InterruptedException {
@Test
public void updateAccessAnotherAclYamlConfigTest() throws IOException, InterruptedException {
- String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl_test.yml";
+ String fileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/acl/plain_acl_test.yml".replace("/", File.separator);
File transport = new File(fileName);
transport.delete();
transport.createNewFile();
@@ -743,8 +816,12 @@ public void updateAccessAnotherAclYamlConfigTest() throws IOException, Interrupt
@Test(expected = AclException.class)
public void createAndUpdateAccessAclNullSkExceptionTest() {
- String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl.yml";
- Map backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+ String backupFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl_bak.yml".replace("/", File.separator);
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(backupFileName, PlainAccessData.class);
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
plainAccessConfig.setAccessKey("RocketMQ33");
@@ -759,8 +836,12 @@ public void createAndUpdateAccessAclNullSkExceptionTest() {
@Test
public void addAccessDefaultAclYamlConfigTest() throws InterruptedException {
PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
- String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+ String backupFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl_bak.yml".replace("/", File.separator);
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(backupFileName, PlainAccessData.class);
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
plainAccessConfig.setAccessKey("watchrocketmqh");
@@ -787,16 +868,17 @@ public void addAccessDefaultAclYamlConfigTest() throws InterruptedException {
Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_WHITE_ADDR), "127.0.0.1");
Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_ADMIN_ROLE), false);
- Map readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
- List> dataVersions = (List>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
- Assert.assertEquals(1, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
+ PlainAccessData readableMap = AclUtils.getYamlDataObject(targetFileName, PlainAccessData.class);
+ List dataVersions = readableMap.getDataVersion();
+ Assert.assertEquals(1L, dataVersions.get(0).getCounter());
AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
}
@Test
public void deleteAccessAnotherAclYamlConfigTest() throws IOException, InterruptedException {
- String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl_test.yml";
+ String fileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/acl/plain_acl_test.yml".replace("/", File.separator);
File transport = new File(fileName);
transport.delete();
transport.createNewFile();
@@ -849,8 +931,12 @@ public void getAllAclConfigTest() {
@Test
public void updateAccessConfigEmptyPermListTest() {
- String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+ String backupFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl_bak.yml".replace("/", File.separator);
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(backupFileName, PlainAccessData.class);
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
@@ -877,8 +963,12 @@ public void updateAccessConfigEmptyPermListTest() {
@Test
public void updateAccessConfigEmptyWhiteRemoteAddressTest() {
- String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
- Map backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+ String backupFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl_bak.yml".replace("/", File.separator);
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/plain_acl.yml".replace("/", File.separator);
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(backupFileName, PlainAccessData.class);
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
@@ -902,4 +992,121 @@ public void updateAccessConfigEmptyWhiteRemoteAddressTest() {
plainAccessValidator.deleteAccessConfig(accessKey);
AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
}
+
+ @Test
+ public void deleteAccessAclToEmptyTest() {
+ final String bakAclFileProp = System.getProperty("rocketmq.acl.plain.file");
+ System.setProperty("rocketmq.acl.plain.file", "conf/empty.yml".replace("/", File.separator));
+ PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
+ plainAccessConfig.setAccessKey("deleteAccessAclToEmpty");
+ plainAccessConfig.setSecretKey("12345678");
+
+ PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+ plainAccessValidator.updateAccessConfig(plainAccessConfig);
+ boolean success = plainAccessValidator.deleteAccessConfig("deleteAccessAclToEmpty");
+ if (null != bakAclFileProp) {
+ System.setProperty("rocketmq.acl.plain.file", bakAclFileProp);
+ } else {
+ System.clearProperty("rocketmq.acl.plain.file");
+ }
+ Assert.assertTrue(success);
+ }
+
+ @Test
+ public void testValidateAfterUpdateAccessConfig() throws NoSuchFieldException, IllegalAccessException {
+ String targetFileName = System.getProperty("rocketmq.home.dir")
+ + File.separator + "conf/update.yml".replace("/", File.separator);
+ System.setProperty("rocketmq.acl.plain.file", "conf/update.yml".replace("/", File.separator));
+ PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+ PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
+ String accessKey = "updateAccessConfig";
+ String secretKey = "123456789111";
+ plainAccessConfig.setAccessKey(accessKey);
+ plainAccessConfig.setSecretKey(secretKey);
+ plainAccessConfig.setAdmin(true);
+ // update
+ plainAccessValidator.updateAccessConfig(plainAccessConfig);
+ // call load
+ Class clazz = PlainAccessValidator.class;
+ Field f = clazz.getDeclaredField("aclPlugEngine");
+ f.setAccessible(true);
+ PlainPermissionManager aclPlugEngine = (PlainPermissionManager) f.get(plainAccessValidator);
+ aclPlugEngine.load(targetFileName);
+
+ // call validate
+ PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
+ pullMessageRequestHeader.setTopic("topicC");
+ pullMessageRequestHeader.setConsumerGroup("consumerGroupA");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
+
+ AclClientRPCHook aclClient = new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
+ aclClient.doBeforeRequest("", remotingCommand);
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ try {
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "1.1.1.1:9876");
+ plainAccessValidator.validate(accessResource);
+ } catch (RemotingCommandException e) {
+ e.printStackTrace();
+ Assert.fail("Should not throw IOException");
+ } finally {
+ System.setProperty("rocketmq.acl.plain.file", "conf/plain_acl.yml".replace("/", File.separator));
+ }
+ }
+
+ /**
+ * Fixme: this test case is not thread safe. The design itself is buggy.
+ * @throws IOException
+ */
+ @Test
+ public void testUpdateSpecifiedAclFileGlobalWhiteAddrsConfig() throws IOException {
+ String folder = "update_global_white_addr";
+ File home = AclTestHelper.copyResources(folder);
+ System.setProperty("rocketmq.home.dir", home.getAbsolutePath());
+ System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml".replace("/", File.separator));
+
+ String targetFileName = Joiner.on(File.separator).join(new String[]{home.getAbsolutePath(), "conf", "plain_acl.yml"});
+ PlainAccessData backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, PlainAccessData.class);
+
+ String targetFileName1 = Joiner.on(File.separator).join(new String[]{home.getAbsolutePath(), "conf", "acl", "plain_acl.yml"});
+ PlainAccessData backUpAclConfigMap1 = AclUtils.getYamlDataObject(targetFileName1, PlainAccessData.class);
+
+ String targetFileName2 = Joiner.on(File.separator).join(new String[]{home.getAbsolutePath(), "conf", "acl", "empty.yml"});
+ PlainAccessData backUpAclConfigMap2 = AclUtils.getYamlDataObject(targetFileName2, PlainAccessData.class);
+
+ PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+ List globalWhiteAddrsList1 = new ArrayList<>();
+ globalWhiteAddrsList1.add("10.10.154.1");
+ List globalWhiteAddrsList2 = new ArrayList<>();
+ globalWhiteAddrsList2.add("10.10.154.2");
+ List globalWhiteAddrsList3 = new ArrayList<>();
+ globalWhiteAddrsList3.add("10.10.154.3");
+
+ //Test parameter p is null
+ plainAccessValidator.updateGlobalWhiteAddrsConfig(globalWhiteAddrsList1, null);
+ String defaultAclFile = targetFileName;
+ PlainAccessData defaultAclFileMap = AclUtils.getYamlDataObject(defaultAclFile, PlainAccessData.class);
+ List defaultAclFileGlobalWhiteAddrList = defaultAclFileMap.getGlobalWhiteRemoteAddresses();
+ Assert.assertTrue(defaultAclFileGlobalWhiteAddrList.contains("10.10.154.1"));
+ //Test parameter p is not null
+ plainAccessValidator.updateGlobalWhiteAddrsConfig(globalWhiteAddrsList2, targetFileName1);
+ PlainAccessData aclFileMap1 = AclUtils.getYamlDataObject(targetFileName1, PlainAccessData.class);
+ List aclFileGlobalWhiteAddrList1 = aclFileMap1.getGlobalWhiteRemoteAddresses();
+ Assert.assertTrue(aclFileGlobalWhiteAddrList1.contains("10.10.154.2"));
+ //Test parameter p is not null, but the file does not have globalWhiteRemoteAddresses
+ plainAccessValidator.updateGlobalWhiteAddrsConfig(globalWhiteAddrsList3, targetFileName2);
+ PlainAccessData aclFileMap2 = AclUtils.getYamlDataObject(targetFileName2, PlainAccessData.class);
+ List aclFileGlobalWhiteAddrList2 = aclFileMap2.getGlobalWhiteRemoteAddresses();
+ Assert.assertTrue(aclFileGlobalWhiteAddrList2.contains("10.10.154.3"));
+
+ AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
+ AclUtils.writeDataObject(targetFileName1, backUpAclConfigMap1);
+ AclUtils.writeDataObject(targetFileName2, backUpAclConfigMap2);
+
+ AclTestHelper.recursiveDelete(home);
+ }
+
+
}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionManagerTest.java b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionManagerTest.java
index d2dabc0b2e7..941d8c77923 100644
--- a/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionManagerTest.java
+++ b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionManagerTest.java
@@ -16,34 +16,46 @@
*/
package org.apache.rocketmq.acl.plain;
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import com.google.common.base.Joiner;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.acl.common.AclConstants;
import org.apache.rocketmq.acl.common.AclException;
import org.apache.rocketmq.acl.common.AclUtils;
import org.apache.rocketmq.acl.common.Permission;
+import org.apache.rocketmq.common.AclConfig;
import org.apache.rocketmq.common.PlainAccessConfig;
+import org.apache.rocketmq.remoting.protocol.DataVersion;
+import org.assertj.core.api.Assertions;
+import org.assertj.core.util.Lists;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
public class PlainPermissionManagerTest {
PlainPermissionManager plainPermissionManager;
- PlainAccessResource PUBPlainAccessResource;
- PlainAccessResource SUBPlainAccessResource;
- PlainAccessResource ANYPlainAccessResource;
- PlainAccessResource DENYPlainAccessResource;
+ PlainAccessResource pubPlainAccessResource;
+ PlainAccessResource subPlainAccessResource;
+ PlainAccessResource anyPlainAccessResource;
+ PlainAccessResource denyPlainAccessResource;
PlainAccessResource plainAccessResource = new PlainAccessResource();
PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
Set adminCode = new HashSet<>();
+ private static final String DEFAULT_TOPIC = "topic-acl";
+
+ private File confHome;
+
@Before
public void init() throws NoSuchFieldException, SecurityException, IOException {
// UPDATE_AND_CREATE_TOPIC
@@ -57,16 +69,15 @@ public void init() throws NoSuchFieldException, SecurityException, IOException {
// DELETE_SUBSCRIPTIONGROUP
adminCode.add(207);
- PUBPlainAccessResource = clonePlainAccessResource(Permission.PUB);
- SUBPlainAccessResource = clonePlainAccessResource(Permission.SUB);
- ANYPlainAccessResource = clonePlainAccessResource(Permission.ANY);
- DENYPlainAccessResource = clonePlainAccessResource(Permission.DENY);
-
- File file = new File("src/test/resources");
- System.setProperty("rocketmq.home.dir", file.getAbsolutePath());
+ pubPlainAccessResource = clonePlainAccessResource(Permission.PUB);
+ subPlainAccessResource = clonePlainAccessResource(Permission.SUB);
+ anyPlainAccessResource = clonePlainAccessResource(Permission.ANY);
+ denyPlainAccessResource = clonePlainAccessResource(Permission.DENY);
+ String folder = "conf";
+ confHome = AclTestHelper.copyResources(folder, true);
+ System.setProperty("rocketmq.home.dir", confHome.getAbsolutePath());
plainPermissionManager = new PlainPermissionManager();
-
}
public PlainAccessResource clonePlainAccessResource(byte perm) {
@@ -107,7 +118,7 @@ public void buildPlainAccessResourceTest() {
plainAccessResource = plainPermissionManager.buildPlainAccessResource(plainAccess);
Assert.assertEquals(plainAccessResource.isAdmin(), true);
- List groups = new ArrayList();
+ List groups = new ArrayList<>();
groups.add("groupA=DENY");
groups.add("groupB=PUB|SUB");
groups.add("groupC=PUB");
@@ -120,7 +131,7 @@ public void buildPlainAccessResourceTest() {
Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupB")).byteValue(), Permission.PUB | Permission.SUB);
Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupC")).byteValue(), Permission.PUB);
- List topics = new ArrayList();
+ List topics = new ArrayList<>();
topics.add("topicA=DENY");
topics.add("topicB=PUB|SUB");
topics.add("topicC=PUB");
@@ -138,7 +149,7 @@ public void buildPlainAccessResourceTest() {
public void checkPermAdmin() {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setRequestCode(17);
- plainPermissionManager.checkPerm(plainAccessResource, PUBPlainAccessResource);
+ plainPermissionManager.checkPerm(plainAccessResource, pubPlainAccessResource);
}
@Test
@@ -146,15 +157,15 @@ public void checkPerm() {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.addResourceAndPerm("topicA", Permission.PUB);
- plainPermissionManager.checkPerm(plainAccessResource, PUBPlainAccessResource);
+ plainPermissionManager.checkPerm(plainAccessResource, pubPlainAccessResource);
plainAccessResource.addResourceAndPerm("topicB", Permission.SUB);
- plainPermissionManager.checkPerm(plainAccessResource, ANYPlainAccessResource);
+ plainPermissionManager.checkPerm(plainAccessResource, anyPlainAccessResource);
plainAccessResource = new PlainAccessResource();
plainAccessResource.addResourceAndPerm("topicB", Permission.SUB);
- plainPermissionManager.checkPerm(plainAccessResource, SUBPlainAccessResource);
+ plainPermissionManager.checkPerm(plainAccessResource, subPlainAccessResource);
plainAccessResource.addResourceAndPerm("topicA", Permission.PUB);
- plainPermissionManager.checkPerm(plainAccessResource, ANYPlainAccessResource);
+ plainPermissionManager.checkPerm(plainAccessResource, anyPlainAccessResource);
}
@@ -163,7 +174,7 @@ public void checkErrorPermDefaultValueNotMatch() {
plainAccessResource = new PlainAccessResource();
plainAccessResource.addResourceAndPerm("topicF", Permission.PUB);
- plainPermissionManager.checkPerm(plainAccessResource, SUBPlainAccessResource);
+ plainPermissionManager.checkPerm(plainAccessResource, subPlainAccessResource);
}
@Test(expected = AclException.class)
@@ -190,7 +201,6 @@ public void passWordThanTest() {
plainPermissionManager.buildPlainAccessResource(plainAccessConfig);
}
-
@SuppressWarnings("unchecked")
@Test
public void cleanAuthenticationInfoTest() throws IllegalAccessException {
@@ -201,7 +211,6 @@ public void cleanAuthenticationInfoTest() throws IllegalAccessException {
plainPermissionManager.clearPermissionInfo();
plainAccessResourceMap = (Map>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
Assert.assertTrue(plainAccessResourceMap.isEmpty());
- // RemoveDataVersionFromYamlFile("src/test/resources/conf/plain_acl.yml");
}
@Test
@@ -209,15 +218,11 @@ public void isWatchStartTest() {
PlainPermissionManager plainPermissionManager = new PlainPermissionManager();
Assert.assertTrue(plainPermissionManager.isWatchStart());
- // RemoveDataVersionFromYamlFile("src/test/resources/conf/plain_acl.yml");
}
@Test
public void testWatch() throws IOException, IllegalAccessException, InterruptedException {
- File file = new File("src/test/resources");
- System.setProperty("rocketmq.home.dir", file.getAbsolutePath());
-
- String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "/conf/acl/plain_acl_test.yml";
+ String fileName = Joiner.on(File.separator).join(new String[]{System.getProperty("rocketmq.home.dir"), "conf", "acl", "plain_acl_test.yml"});
File transport = new File(fileName);
transport.delete();
transport.createNewFile();
@@ -246,13 +251,11 @@ public void testWatch() throws IOException, IllegalAccessException, InterruptedE
}
- Map updatedMap = AclUtils.getYamlDataObject(fileName, Map.class);
- List> accounts = (List>) updatedMap.get("accounts");
- accounts.get(0).remove("accessKey");
- accounts.get(0).remove("secretKey");
- accounts.get(0).put("accessKey", "watchrocketmq1y");
- accounts.get(0).put("secretKey", "88888888");
- accounts.get(0).put("admin", "false");
+ PlainAccessData updatedMap = AclUtils.getYamlDataObject(fileName, PlainAccessData.class);
+ List accounts = updatedMap.getAccounts();
+ accounts.get(0).setAccessKey("watchrocketmq1y");
+ accounts.get(0).setSecretKey("88888888");
+ accounts.get(0).setAdmin(false);
// Update file and flush to yaml file
AclUtils.writeDataObject(fileName, updatedMap);
@@ -266,6 +269,111 @@ public void testWatch() throws IOException, IllegalAccessException, InterruptedE
}
transport.delete();
- System.setProperty("rocketmq.home.dir", "src/test/resources");
}
+
+ @Test
+ public void updateAccessConfigTest() {
+ Assert.assertThrows(AclException.class, () -> plainPermissionManager.updateAccessConfig(null));
+
+ plainAccessConfig.setAccessKey("admin_test");
+ // Invalid parameter
+ plainAccessConfig.setSecretKey("123456");
+ plainAccessConfig.setAdmin(true);
+ Assert.assertThrows(AclException.class, () -> plainPermissionManager.updateAccessConfig(plainAccessConfig));
+
+ plainAccessConfig.setSecretKey("12345678");
+ // Invalid parameter
+ plainAccessConfig.setGroupPerms(Lists.newArrayList("groupA!SUB"));
+ Assert.assertThrows(AclException.class, () -> plainPermissionManager.updateAccessConfig(plainAccessConfig));
+
+ // first update
+ plainAccessConfig.setGroupPerms(Lists.newArrayList("groupA=SUB"));
+ plainPermissionManager.updateAccessConfig(plainAccessConfig);
+
+ // second update
+ plainAccessConfig.setTopicPerms(Lists.newArrayList("topicA=SUB"));
+ plainPermissionManager.updateAccessConfig(plainAccessConfig);
+ }
+
+ @Test
+ public void getAllAclFilesTest() {
+ final List notExistList = plainPermissionManager.getAllAclFiles("aa/bb");
+ Assertions.assertThat(notExistList).isEmpty();
+ final List files = plainPermissionManager.getAllAclFiles(confHome.getAbsolutePath());
+ Assertions.assertThat(files).isNotEmpty();
+ }
+
+ @Test
+ public void loadTest() {
+ plainPermissionManager.load();
+ final Map map = plainPermissionManager.getDataVersionMap();
+ Assertions.assertThat(map).isNotEmpty();
+ }
+
+ @Test
+ public void updateAclConfigFileVersionTest() {
+ String aclFileName = "test_plain_acl";
+ PlainAccessData updateAclConfigMap = new PlainAccessData();
+ List versionElement = new ArrayList<>();
+ PlainAccessData.DataVersion accountsMap = new PlainAccessData.DataVersion();
+ accountsMap.setCounter(1);
+ accountsMap.setTimestamp(System.currentTimeMillis());
+ versionElement.add(accountsMap);
+
+ updateAclConfigMap.setDataVersion(versionElement);
+ final PlainAccessData map = plainPermissionManager.updateAclConfigFileVersion(aclFileName, updateAclConfigMap);
+ final List version = map.getDataVersion();
+ Assert.assertEquals(2L, version.get(0).getCounter());
+ }
+
+ @Test
+ public void createAclAccessConfigMapTest() {
+ PlainAccessConfig existedAccountMap = new PlainAccessConfig();
+ plainAccessConfig.setAccessKey("admin123");
+ plainAccessConfig.setSecretKey("12345678");
+ plainAccessConfig.setWhiteRemoteAddress("192.168.1.1");
+ plainAccessConfig.setAdmin(false);
+ plainAccessConfig.setDefaultGroupPerm(AclConstants.SUB_PUB);
+ plainAccessConfig.setTopicPerms(Arrays.asList(DEFAULT_TOPIC + "=" + AclConstants.PUB));
+ plainAccessConfig.setGroupPerms(Lists.newArrayList("groupA=SUB"));
+
+ final PlainAccessConfig map = plainPermissionManager.createAclAccessConfigMap(existedAccountMap, plainAccessConfig);
+ Assert.assertEquals(AclConstants.SUB_PUB, map.getDefaultGroupPerm());
+ Assert.assertEquals("groupA=SUB", map.getGroupPerms().get(0));
+ Assert.assertEquals("12345678", map.getSecretKey());
+ Assert.assertEquals("admin123", map.getAccessKey());
+ Assert.assertEquals("192.168.1.1", map.getWhiteRemoteAddress());
+ Assert.assertEquals("topic-acl=PUB", map.getTopicPerms().get(0));
+ Assert.assertEquals(false, map.isAdmin());
+ }
+
+ @Test
+ public void deleteAccessConfigTest() throws InterruptedException {
+ // delete not exist accessConfig
+ final boolean flag1 = plainPermissionManager.deleteAccessConfig("test_delete");
+ assert !flag1;
+
+ plainAccessConfig.setAccessKey("test_delete");
+ plainAccessConfig.setSecretKey("12345678");
+ plainAccessConfig.setWhiteRemoteAddress("192.168.1.1");
+ plainAccessConfig.setAdmin(false);
+ plainAccessConfig.setDefaultGroupPerm(AclConstants.SUB_PUB);
+ plainAccessConfig.setTopicPerms(Arrays.asList(DEFAULT_TOPIC + "=" + AclConstants.PUB));
+ plainAccessConfig.setGroupPerms(Lists.newArrayList("groupA=SUB"));
+ plainPermissionManager.updateAccessConfig(plainAccessConfig);
+
+ //delete existed accessConfig
+ final boolean flag2 = plainPermissionManager.deleteAccessConfig("test_delete");
+ assert flag2;
+
+ }
+
+ @Test
+ public void updateGlobalWhiteAddrsConfigTest() {
+ final boolean flag = plainPermissionManager.updateGlobalWhiteAddrsConfig(Lists.newArrayList("192.168.1.2"));
+ assert flag;
+ final AclConfig config = plainPermissionManager.getAllAclConfig();
+ Assert.assertEquals(true, config.getGlobalWhiteAddrs().contains("192.168.1.2"));
+ }
+
}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyTest.java b/acl/src/test/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyTest.java
index 87eb37bdf2f..df7dd0c5460 100644
--- a/acl/src/test/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyTest.java
+++ b/acl/src/test/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyTest.java
@@ -25,7 +25,7 @@ public class RemoteAddressStrategyTest {
RemoteAddressStrategyFactory remoteAddressStrategyFactory = new RemoteAddressStrategyFactory();
@Test
- public void netaddressStrategyFactoryExceptionTest() {
+ public void netAddressStrategyFactoryExceptionTest() {
PlainAccessResource plainAccessResource = new PlainAccessResource();
remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
Assert.assertEquals(remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource).getClass(),
@@ -33,7 +33,7 @@ public void netaddressStrategyFactoryExceptionTest() {
}
@Test
- public void netaddressStrategyFactoryTest() {
+ public void netAddressStrategyFactoryTest() {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("*");
@@ -114,19 +114,19 @@ public void verifyTest() {
}
@Test
- public void nullNetaddressStrategyTest() {
+ public void nullNetAddressStrategyTest() {
boolean isMatch = RemoteAddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY.match(new PlainAccessResource());
Assert.assertTrue(isMatch);
}
@Test
- public void blankNetaddressStrategyTest() {
+ public void blankNetAddressStrategyTest() {
boolean isMatch = RemoteAddressStrategyFactory.BLANK_NET_ADDRESS_STRATEGY.match(new PlainAccessResource());
Assert.assertFalse(isMatch);
}
@Test
- public void oneNetaddressStrategyTest() {
+ public void oneNetAddressStrategyTest() {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
RemoteAddressStrategy remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
@@ -164,15 +164,15 @@ public void oneNetaddressStrategyTest() {
}
@Test
- public void multipleNetaddressStrategyTest() {
+ public void multipleNetAddressStrategyTest() {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("127.0.0.1,127.0.0.2,127.0.0.3");
RemoteAddressStrategy remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- multipleNetaddressStrategyTest(remoteAddressStrategy);
+ multipleNetAddressStrategyTest(remoteAddressStrategy);
plainAccessResource.setWhiteRemoteAddress("127.0.0.{1,2,3}");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- multipleNetaddressStrategyTest(remoteAddressStrategy);
+ multipleNetAddressStrategyTest(remoteAddressStrategy);
plainAccessResource.setWhiteRemoteAddress("192.100-150.*.*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
@@ -183,16 +183,16 @@ public void multipleNetaddressStrategyTest() {
plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:1,1050::0005:0600:300c:2,1050::0005:0600:300c:3");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- multipleIPv6NetaddressStrategyTest(remoteAddressStrategy);
+ multipleIPv6NetAddressStrategyTest(remoteAddressStrategy);
plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:{1,2,3}");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- multipleIPv6NetaddressStrategyTest(remoteAddressStrategy);
+ multipleIPv6NetAddressStrategyTest(remoteAddressStrategy);
}
@Test(expected = AclException.class)
- public void multipleNetaddressStrategyExceptionTest() {
+ public void multipleNetAddressStrategyExceptionTest() {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("127.0.0.1,2,3}");
remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
@@ -210,7 +210,7 @@ public void multipleNetaddressStrategyExceptionTest() {
remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
}
- private void multipleNetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy) {
+ private void multipleNetAddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy) {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
boolean match = remoteAddressStrategy.match(plainAccessResource);
@@ -234,7 +234,7 @@ private void multipleNetaddressStrategyTest(RemoteAddressStrategy remoteAddressS
}
- private void multipleIPv6NetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy) {
+ private void multipleIPv6NetAddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy) {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("1050:0000:0000:0000:0005:0600:300c:1");
boolean match = remoteAddressStrategy.match(plainAccessResource);
@@ -259,57 +259,57 @@ private void multipleIPv6NetaddressStrategyTest(RemoteAddressStrategy remoteAddr
}
@Test
- public void rangeNetaddressStrategyTest() {
+ public void rangeNetAddressStrategyTest() {
String head = "127.0.0.";
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("127.0.0.1-200");
RemoteAddressStrategy remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeNetaddressStrategyTest(remoteAddressStrategy, head, 1, 200, true);
+ rangeNetAddressStrategyTest(remoteAddressStrategy, head, 1, 200, true);
plainAccessResource.setWhiteRemoteAddress("127.0.0.*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeNetaddressStrategyTest(remoteAddressStrategy, head, 0, 255, true);
+ rangeNetAddressStrategyTest(remoteAddressStrategy, head, 0, 255, true);
plainAccessResource.setWhiteRemoteAddress("127.0.1-200.*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeNetaddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
+ rangeNetAddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
plainAccessResource.setWhiteRemoteAddress("127.*.*.*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeNetaddressStrategyTest(remoteAddressStrategy, head, 0, 255, true);
+ rangeNetAddressStrategyTest(remoteAddressStrategy, head, 0, 255, true);
plainAccessResource.setWhiteRemoteAddress("127.1-150.*.*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeNetaddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
+ rangeNetAddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
// IPv6 test
head = "1050::0005:0600:300c:";
plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:1-200");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "1", "200", true);
+ rangeIPv6NetAddressStrategyTest(remoteAddressStrategy, head, "1", "200", true);
plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", true);
+ rangeIPv6NetAddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", true);
plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:3001:*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", false);
+ rangeIPv6NetAddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", false);
head = "1050::0005:0600:300c:1:";
plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:1-200:*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", true);
+ rangeIPv6NetAddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", true);
head = "1050::0005:0600:300c:201:";
plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:1-200:*");
remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", false);
+ rangeIPv6NetAddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", false);
}
- private void rangeNetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy, String head, int start,
+ private void rangeNetAddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy, String head, int start,
int end,
boolean isFalse) {
PlainAccessResource plainAccessResource = new PlainAccessResource();
@@ -325,18 +325,18 @@ private void rangeNetaddressStrategyTest(RemoteAddressStrategy remoteAddressStra
}
}
- private void rangeNetaddressStrategyThirdlyTest(RemoteAddressStrategy remoteAddressStrategy, String head, int start,
+ private void rangeNetAddressStrategyThirdlyTest(RemoteAddressStrategy remoteAddressStrategy, String head, int start,
int end) {
String newHead;
for (int i = -10; i < 300; i++) {
newHead = head + i;
if (i >= start && i <= end) {
- rangeNetaddressStrategyTest(remoteAddressStrategy, newHead, 0, 255, false);
+ rangeNetAddressStrategyTest(remoteAddressStrategy, newHead, 0, 255, false);
}
}
}
- private void rangeIPv6NetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy, String head, String start,
+ private void rangeIPv6NetAddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy, String head, String start,
String end,
boolean isFalse) {
PlainAccessResource plainAccessResource = new PlainAccessResource();
@@ -356,23 +356,23 @@ private void rangeIPv6NetaddressStrategyTest(RemoteAddressStrategy remoteAddress
}
@Test(expected = AclException.class)
- public void rangeNetaddressStrategyExceptionStartGreaterEndTest() {
- rangeNetaddressStrategyExceptionTest("127.0.0.2-1");
+ public void rangeNetAddressStrategyExceptionStartGreaterEndTest() {
+ rangeNetAddressStrategyExceptionTest("127.0.0.2-1");
}
@Test(expected = AclException.class)
- public void rangeNetaddressStrategyExceptionScopeTest() {
- rangeNetaddressStrategyExceptionTest("127.0.0.-1-200");
+ public void rangeNetAddressStrategyExceptionScopeTest() {
+ rangeNetAddressStrategyExceptionTest("127.0.0.-1-200");
}
@Test(expected = AclException.class)
- public void rangeNetaddressStrategyExceptionScopeTwoTest() {
- rangeNetaddressStrategyExceptionTest("127.0.0.0-256");
+ public void rangeNetAddressStrategyExceptionScopeTwoTest() {
+ rangeNetAddressStrategyExceptionTest("127.0.0.0-256");
}
- private void rangeNetaddressStrategyExceptionTest(String netaddress) {
+ private void rangeNetAddressStrategyExceptionTest(String netAddress) {
PlainAccessResource plainAccessResource = new PlainAccessResource();
- plainAccessResource.setWhiteRemoteAddress(netaddress);
+ plainAccessResource.setWhiteRemoteAddress(netAddress);
remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
}
diff --git a/acl/src/test/resources/both_acl_file_folder_conf/conf/acl/plain_acl.yml b/acl/src/test/resources/both_acl_file_folder_conf/conf/acl/plain_acl.yml
new file mode 100644
index 00000000000..cf4ea7f4a5b
--- /dev/null
+++ b/acl/src/test/resources/both_acl_file_folder_conf/conf/acl/plain_acl.yml
@@ -0,0 +1,39 @@
+# 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.
+
+## no global white addresses in this file, define them in ../plain_acl.yml
+accounts:
+ - accessKey: RocketMQ
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.0.*
+ admin: false
+ defaultTopicPerm: DENY
+ defaultGroupPerm: SUB
+ topicPerms:
+ - topicA=DENY
+ - topicB=PUB|SUB
+ - topicC=SUB
+ groupPerms:
+ # the group should convert to retry topic
+ - groupA=DENY
+ - groupB=SUB
+ - groupC=SUB
+
+ - accessKey: rocketmq2
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.1.*
+ # if it is admin, it could access all resources
+ admin: true
+
diff --git a/acl/src/test/resources/both_acl_file_folder_conf/conf/plain_acl.yml b/acl/src/test/resources/both_acl_file_folder_conf/conf/plain_acl.yml
new file mode 100644
index 00000000000..41afea097ae
--- /dev/null
+++ b/acl/src/test/resources/both_acl_file_folder_conf/conf/plain_acl.yml
@@ -0,0 +1,21 @@
+# 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.
+
+## suggested format
+
+globalWhiteRemoteAddresses:
+ - 10.10.103.*
+ - 192.168.0.*
+
diff --git a/acl/src/test/resources/conf/plain_acl.yml b/acl/src/test/resources/conf/plain_acl.yml
index 2c24795ff64..59bd6d4ff29 100644
--- a/acl/src/test/resources/conf/plain_acl.yml
+++ b/acl/src/test/resources/conf/plain_acl.yml
@@ -18,7 +18,6 @@
globalWhiteRemoteAddresses:
- 10.10.103.*
- 192.168.0.*
-
accounts:
- accessKey: RocketMQ
secretKey: 12345678
@@ -31,14 +30,10 @@ accounts:
- topicB=PUB|SUB
- topicC=SUB
groupPerms:
- # the group should convert to retry topic
- groupA=DENY
- groupB=SUB
- groupC=SUB
-
- accessKey: rocketmq2
secretKey: 12345678
whiteRemoteAddress: 192.168.1.*
- # if it is admin, it could access all resources
admin: true
-
diff --git a/acl/src/test/resources/conf/plain_acl_bak.yml b/acl/src/test/resources/conf/plain_acl_bak.yml
new file mode 100644
index 00000000000..59bd6d4ff29
--- /dev/null
+++ b/acl/src/test/resources/conf/plain_acl_bak.yml
@@ -0,0 +1,39 @@
+# 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.
+
+## suggested format
+
+globalWhiteRemoteAddresses:
+- 10.10.103.*
+- 192.168.0.*
+accounts:
+- accessKey: RocketMQ
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.0.*
+ admin: false
+ defaultTopicPerm: DENY
+ defaultGroupPerm: SUB
+ topicPerms:
+ - topicA=DENY
+ - topicB=PUB|SUB
+ - topicC=SUB
+ groupPerms:
+ - groupA=DENY
+ - groupB=SUB
+ - groupC=SUB
+- accessKey: rocketmq2
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.1.*
+ admin: true
diff --git a/acl/src/test/resources/empty_acl_folder_conf/conf/plain_acl.yml b/acl/src/test/resources/empty_acl_folder_conf/conf/plain_acl.yml
new file mode 100644
index 00000000000..6ade46723b7
--- /dev/null
+++ b/acl/src/test/resources/empty_acl_folder_conf/conf/plain_acl.yml
@@ -0,0 +1,19 @@
+# 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.
+
+
+globalWhiteRemoteAddresses:
+ - 10.10.103.*
+ - 192.168.0.*
diff --git a/acl/src/test/resources/only_acl_folder_conf/conf/acl/plain_acl.yml b/acl/src/test/resources/only_acl_folder_conf/conf/acl/plain_acl.yml
new file mode 100644
index 00000000000..cf4ea7f4a5b
--- /dev/null
+++ b/acl/src/test/resources/only_acl_folder_conf/conf/acl/plain_acl.yml
@@ -0,0 +1,39 @@
+# 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.
+
+## no global white addresses in this file, define them in ../plain_acl.yml
+accounts:
+ - accessKey: RocketMQ
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.0.*
+ admin: false
+ defaultTopicPerm: DENY
+ defaultGroupPerm: SUB
+ topicPerms:
+ - topicA=DENY
+ - topicB=PUB|SUB
+ - topicC=SUB
+ groupPerms:
+ # the group should convert to retry topic
+ - groupA=DENY
+ - groupB=SUB
+ - groupC=SUB
+
+ - accessKey: rocketmq2
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.1.*
+ # if it is admin, it could access all resources
+ admin: true
+
diff --git a/acl/src/test/resources/logback-test.xml b/acl/src/test/resources/rmq.logback-test.xml
similarity index 66%
rename from acl/src/test/resources/logback-test.xml
rename to acl/src/test/resources/rmq.logback-test.xml
index e556c649e52..8695d52d57c 100644
--- a/acl/src/test/resources/logback-test.xml
+++ b/acl/src/test/resources/rmq.logback-test.xml
@@ -17,18 +17,20 @@
-->
-
-
- %d{HH:mm:ss.SSS} [%thread] %-5level %logger{5} - %msg%n
- UTF-8
-
+
+
+
+ %d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n
+
+
-
-
+
+
-
-
+
+
+
-
+
\ No newline at end of file
diff --git a/acl/src/test/resources/update_global_white_addr/conf/acl/empty.yml b/acl/src/test/resources/update_global_white_addr/conf/acl/empty.yml
new file mode 100644
index 00000000000..52ff50c2bef
--- /dev/null
+++ b/acl/src/test/resources/update_global_white_addr/conf/acl/empty.yml
@@ -0,0 +1,18 @@
+# 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.
+
+## suggested format
+
+accounts: []
diff --git a/acl/src/test/resources/update_global_white_addr/conf/acl/plain_acl.yml b/acl/src/test/resources/update_global_white_addr/conf/acl/plain_acl.yml
new file mode 100644
index 00000000000..59bd6d4ff29
--- /dev/null
+++ b/acl/src/test/resources/update_global_white_addr/conf/acl/plain_acl.yml
@@ -0,0 +1,39 @@
+# 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.
+
+## suggested format
+
+globalWhiteRemoteAddresses:
+- 10.10.103.*
+- 192.168.0.*
+accounts:
+- accessKey: RocketMQ
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.0.*
+ admin: false
+ defaultTopicPerm: DENY
+ defaultGroupPerm: SUB
+ topicPerms:
+ - topicA=DENY
+ - topicB=PUB|SUB
+ - topicC=SUB
+ groupPerms:
+ - groupA=DENY
+ - groupB=SUB
+ - groupC=SUB
+- accessKey: rocketmq2
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.1.*
+ admin: true
diff --git a/acl/src/test/resources/update_global_white_addr/conf/plain_acl.yml b/acl/src/test/resources/update_global_white_addr/conf/plain_acl.yml
new file mode 100644
index 00000000000..64c6e34169c
--- /dev/null
+++ b/acl/src/test/resources/update_global_white_addr/conf/plain_acl.yml
@@ -0,0 +1,36 @@
+# 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.
+
+## suggested format
+
+accounts:
+- accessKey: RocketMQ
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.0.*
+ admin: false
+ defaultTopicPerm: DENY
+ defaultGroupPerm: SUB
+ topicPerms:
+ - topicA=DENY
+ - topicB=PUB|SUB
+ - topicC=SUB
+ groupPerms:
+ - groupA=DENY
+ - groupB=SUB
+ - groupC=SUB
+- accessKey: rocketmq2
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.1.*
+ admin: true
diff --git a/bazel/BUILD.bazel b/bazel/BUILD.bazel
new file mode 100644
index 00000000000..a9fd83fea0b
--- /dev/null
+++ b/bazel/BUILD.bazel
@@ -0,0 +1,16 @@
+#
+# 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.
+#
\ No newline at end of file
diff --git a/bazel/GenTestRules.bzl b/bazel/GenTestRules.bzl
new file mode 100644
index 00000000000..73cbb4a84ba
--- /dev/null
+++ b/bazel/GenTestRules.bzl
@@ -0,0 +1,109 @@
+#
+# 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.
+#
+
+"""Generate java test rules from given test_files.
+
+Instead of having to create one test rule per test in the BUILD file, this rule
+provides a handy way to create a bunch of test rules for the specified test
+files.
+
+"""
+
+def GenTestRules(
+ name,
+ test_files,
+ deps,
+ exclude_tests = [],
+ default_test_size = "small",
+ small_tests = [],
+ medium_tests = [],
+ large_tests = [],
+ enormous_tests = [],
+ resources = [],
+ flaky_tests = [],
+ tags = [],
+ prefix = "",
+ jvm_flags = [],
+ args = [],
+ visibility = None,
+ shard_count = 1):
+ for test in _get_test_names(test_files):
+ if test in exclude_tests:
+ continue
+ test_size = default_test_size
+ if test in small_tests:
+ test_size = "small"
+ if test in medium_tests:
+ test_size = "medium"
+ if test in large_tests:
+ test_size = "large"
+ if test in enormous_tests:
+ test_size = "enormous"
+ flaky = 0
+ if (test in flaky_tests) or ("flaky" in tags):
+ flaky = 1
+ java_class = _package_from_path(
+ native.package_name() + "/" + _strip_right(test, ".java"),
+ )
+ package = java_class[:java_class.rfind(".")]
+ native.java_test(
+ name = prefix + test,
+ runtime_deps = deps,
+ resources = resources,
+ size = test_size,
+ jvm_flags = jvm_flags,
+ args = args,
+ flaky = flaky,
+ tags = tags,
+ test_class = java_class,
+ visibility = visibility,
+ shard_count = shard_count,
+ )
+
+def _get_test_names(test_files):
+ test_names = []
+ for test_file in test_files:
+ if not test_file.endswith("Test.java") and not test_file.endswith("IT.java"):
+ continue
+ test_names += [test_file[:-5]]
+ return test_names
+
+def _package_from_path(package_path, src_impls = None):
+ src_impls = src_impls or ["javatests/", "java/"]
+ for src_impl in src_impls:
+ if not src_impl.endswith("/"):
+ src_impl += "/"
+ index = _index_of_end(package_path, src_impl)
+ if index >= 0:
+ package_path = package_path[index:]
+ break
+ return package_path.replace("/", ".")
+
+def _strip_right(str, suffix):
+ """Returns str without the suffix if it ends with suffix."""
+ if str.endswith(suffix):
+ return str[0:len(str) - len(suffix)]
+ else:
+ return str
+
+def _index_of_end(str, part):
+ """If part is in str, return the index of the first character after part.
+ Return -1 if part is not in str."""
+ index = str.find(part)
+ if index >= 0:
+ return index + len(part)
+ return -1
diff --git a/broker/BUILD.bazel b/broker/BUILD.bazel
new file mode 100644
index 00000000000..d0d3a2f96ba
--- /dev/null
+++ b/broker/BUILD.bazel
@@ -0,0 +1,93 @@
+#
+# 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.
+#
+load("//bazel:GenTestRules.bzl", "GenTestRules")
+
+java_library(
+ name = "broker",
+ srcs = glob(["src/main/java/**/*.java"]),
+ visibility = ["//visibility:public"],
+ deps = [
+ "//acl",
+ "//client",
+ "//common",
+ "//filter",
+ "//remoting",
+ "//srvutil",
+ "//store",
+ "@maven//:ch_qos_logback_logback_classic",
+ "@maven//:com_alibaba_fastjson",
+ "@maven//:com_github_luben_zstd_jni",
+ "@maven//:com_google_guava_guava",
+ "@maven//:com_googlecode_concurrentlinkedhashmap_concurrentlinkedhashmap_lru",
+ "@maven//:commons_cli_commons_cli",
+ "@maven//:commons_collections_commons_collections",
+ "@maven//:commons_io_commons_io",
+ "@maven//:commons_validator_commons_validator",
+ "@maven//:io_netty_netty_all",
+ "@maven//:io_openmessaging_storage_dledger",
+ "@maven//:io_opentelemetry_opentelemetry_api",
+ "@maven//:io_opentelemetry_opentelemetry_context",
+ "@maven//:io_opentelemetry_opentelemetry_exporter_otlp",
+ "@maven//:io_opentelemetry_opentelemetry_exporter_prometheus",
+ "@maven//:io_opentelemetry_opentelemetry_exporter_logging",
+ "@maven//:io_opentelemetry_opentelemetry_sdk",
+ "@maven//:io_opentelemetry_opentelemetry_sdk_common",
+ "@maven//:io_opentelemetry_opentelemetry_sdk_metrics",
+ "@maven//:org_apache_commons_commons_lang3",
+ "@maven//:org_lz4_lz4_java",
+ "@maven//:io_github_aliyunmq_rocketmq_slf4j_api",
+ "@maven//:io_github_aliyunmq_rocketmq_logback_classic",
+ "@maven//:org_slf4j_jul_to_slf4j",
+ "@maven//:io_github_aliyunmq_rocketmq_shaded_slf4j_api_bridge",
+ ],
+)
+
+java_library(
+ name = "tests",
+ srcs = glob(["src/test/java/**/*.java"]),
+ resources = [
+ "src/test/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator",
+ "src/test/resources/META-INF/service/org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener",
+ "src/test/resources/META-INF/service/org.apache.rocketmq.broker.transaction.TransactionalMessageService",
+ "src/test/resources/rmq.logback-test.xml",
+ ],
+ visibility = ["//visibility:public"],
+ deps = [
+ ":broker",
+ "//:test_deps",
+ "//acl",
+ "//client",
+ "//common",
+ "//filter",
+ "//remoting",
+ "//store",
+ "@maven//:com_alibaba_fastjson",
+ "@maven//:com_google_guava_guava",
+ "@maven//:io_netty_netty_all",
+ "@maven//:org_apache_commons_commons_lang3",
+ "@maven//:io_github_aliyunmq_rocketmq_slf4j_api",
+ "@maven//:org_powermock_powermock_core",
+ ],
+)
+
+GenTestRules(
+ name = "GeneratedTestRules",
+ test_files = glob(["src/test/java/**/*Test.java"]),
+ deps = [
+ ":tests",
+ ],
+)
diff --git a/broker/pom.xml b/broker/pom.xml
index fee6d9ea2c8..70ba0ee665e 100644
--- a/broker/pom.xml
+++ b/broker/pom.xml
@@ -1,19 +1,19 @@
-
org.apache.rocketmq
rocketmq-all
- 4.9.4-SNAPSHOT
+ 5.1.3
4.0.0
@@ -21,10 +21,14 @@
rocketmq-broker
rocketmq-broker ${project.version}
+
+ ${basedir}/..
+
+
${project.groupId}
- rocketmq-common
+ rocketmq-remoting
${project.groupId}
@@ -32,7 +36,15 @@
${project.groupId}
- rocketmq-remoting
+ rocketmq-tiered-store
+
+
+ io.github.aliyunmq
+ rocketmq-slf4j-api
+
+
+ io.github.aliyunmq
+ rocketmq-logback-classic
${project.groupId}
@@ -51,8 +63,8 @@
rocketmq-acl
- ch.qos.logback
- logback-classic
+ commons-io
+ commons-io
com.alibaba
@@ -62,21 +74,29 @@
org.javassist
javassist
-
- org.slf4j
- slf4j-api
-
org.bouncycastle
bcpkix-jdk15on
+
+ com.googlecode.concurrentlinkedhashmap
+ concurrentlinkedhashmap-lru
+
+
+ io.github.aliyunmq
+ rocketmq-shaded-slf4j-api-bridge
+
+
+ org.slf4j
+ jul-to-slf4j
+
maven-surefire-plugin
- 2.19.1
+ ${maven-surefire-plugin.version}
1
false
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
index 7bfc618b922..196401e268b 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -16,24 +16,9 @@
*/
package org.apache.rocketmq.broker;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import com.google.common.collect.Lists;
import org.apache.rocketmq.acl.AccessValidator;
+import org.apache.rocketmq.acl.plain.PlainAccessValidator;
import org.apache.rocketmq.broker.client.ClientHousekeepingService;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ConsumerManager;
@@ -41,56 +26,75 @@
import org.apache.rocketmq.broker.client.ProducerManager;
import org.apache.rocketmq.broker.client.net.Broker2Client;
import org.apache.rocketmq.broker.client.rebalance.RebalanceLockManager;
+import org.apache.rocketmq.broker.coldctr.ColdDataCgCtrService;
+import org.apache.rocketmq.broker.coldctr.ColdDataPullRequestHoldService;
+import org.apache.rocketmq.broker.controller.ReplicasManager;
import org.apache.rocketmq.broker.dledger.DLedgerRoleChangeHandler;
+import org.apache.rocketmq.broker.failover.EscapeBridge;
import org.apache.rocketmq.broker.filter.CommitLogDispatcherCalcBitMap;
import org.apache.rocketmq.broker.filter.ConsumerFilterManager;
-import org.apache.rocketmq.broker.filtersrv.FilterServerManager;
import org.apache.rocketmq.broker.latency.BrokerFastFailure;
import org.apache.rocketmq.broker.latency.BrokerFixedThreadPoolExecutor;
import org.apache.rocketmq.broker.longpolling.LmqPullRequestHoldService;
import org.apache.rocketmq.broker.longpolling.NotifyMessageArrivingListener;
import org.apache.rocketmq.broker.longpolling.PullRequestHoldService;
+import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.broker.mqtrace.ConsumeMessageHook;
import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
+import org.apache.rocketmq.broker.offset.BroadcastOffsetManager;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
+import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.offset.LmqConsumerOffsetManager;
import org.apache.rocketmq.broker.out.BrokerOuterAPI;
-import org.apache.rocketmq.broker.plugin.MessageStoreFactory;
-import org.apache.rocketmq.broker.plugin.MessageStorePluginContext;
+import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin;
+import org.apache.rocketmq.broker.processor.AckMessageProcessor;
import org.apache.rocketmq.broker.processor.AdminBrokerProcessor;
+import org.apache.rocketmq.broker.processor.ChangeInvisibleTimeProcessor;
import org.apache.rocketmq.broker.processor.ClientManageProcessor;
import org.apache.rocketmq.broker.processor.ConsumerManageProcessor;
import org.apache.rocketmq.broker.processor.EndTransactionProcessor;
+import org.apache.rocketmq.broker.processor.NotificationProcessor;
+import org.apache.rocketmq.broker.processor.PeekMessageProcessor;
+import org.apache.rocketmq.broker.processor.PollingInfoProcessor;
+import org.apache.rocketmq.broker.processor.PopInflightMessageCounter;
+import org.apache.rocketmq.broker.processor.PopMessageProcessor;
import org.apache.rocketmq.broker.processor.PullMessageProcessor;
+import org.apache.rocketmq.broker.processor.QueryAssignmentProcessor;
import org.apache.rocketmq.broker.processor.QueryMessageProcessor;
import org.apache.rocketmq.broker.processor.ReplyMessageProcessor;
import org.apache.rocketmq.broker.processor.SendMessageProcessor;
+import org.apache.rocketmq.broker.schedule.ScheduleMessageService;
import org.apache.rocketmq.broker.slave.SlaveSynchronize;
import org.apache.rocketmq.broker.subscription.LmqSubscriptionGroupManager;
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.topic.LmqTopicConfigManager;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
+import org.apache.rocketmq.broker.topic.TopicQueueMappingCleanService;
+import org.apache.rocketmq.broker.topic.TopicQueueMappingManager;
+import org.apache.rocketmq.broker.topic.TopicRouteInfoManager;
import org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener;
import org.apache.rocketmq.broker.transaction.TransactionalMessageCheckService;
import org.apache.rocketmq.broker.transaction.TransactionalMessageService;
import org.apache.rocketmq.broker.transaction.queue.DefaultTransactionalMessageCheckListener;
import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageBridge;
import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageServiceImpl;
-import org.apache.rocketmq.broker.util.ServiceProvider;
+import org.apache.rocketmq.broker.util.HookUtils;
+import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.BrokerConfig;
-import org.apache.rocketmq.common.Configuration;
-import org.apache.rocketmq.common.DataVersion;
+import org.apache.rocketmq.common.BrokerIdentity;
+import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.constant.PermName;
-import org.apache.rocketmq.common.namesrv.RegisterBrokerResult;
-import org.apache.rocketmq.common.protocol.RequestCode;
-import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.stats.MomentStatsItem;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.common.utils.ServiceProvider;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.remoting.Configuration;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.common.TlsMode;
@@ -100,80 +104,190 @@
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.netty.RequestTask;
import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
+import org.apache.rocketmq.remoting.protocol.BrokerSyncInfo;
+import org.apache.rocketmq.remoting.protocol.DataVersion;
+import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.body.BrokerMemberGroup;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigAndMappingSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.namesrv.RegisterBrokerResult;
+import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail;
+import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingInfo;
import org.apache.rocketmq.srvutil.FileWatchService;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.dledger.DLedgerCommitLog;
+import org.apache.rocketmq.store.hook.PutMessageHook;
+import org.apache.rocketmq.store.hook.SendMessageBackHook;
+import org.apache.rocketmq.store.plugin.MessageStoreFactory;
+import org.apache.rocketmq.store.plugin.MessageStorePluginContext;
import org.apache.rocketmq.store.stats.BrokerStats;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.apache.rocketmq.store.stats.LmqBrokerStatsManager;
+import org.apache.rocketmq.store.timer.TimerCheckpoint;
+import org.apache.rocketmq.store.timer.TimerMessageStore;
+import org.apache.rocketmq.store.timer.TimerMetrics;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
public class BrokerController {
- private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
- private static final InternalLogger LOG_PROTECTION = InternalLoggerFactory.getLogger(LoggerName.PROTECTION_LOGGER_NAME);
- private static final InternalLogger LOG_WATER_MARK = InternalLoggerFactory.getLogger(LoggerName.WATER_MARK_LOGGER_NAME);
- private final BrokerConfig brokerConfig;
+ protected static final Logger LOG = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+ private static final Logger LOG_PROTECTION = LoggerFactory.getLogger(LoggerName.PROTECTION_LOGGER_NAME);
+ private static final Logger LOG_WATER_MARK = LoggerFactory.getLogger(LoggerName.WATER_MARK_LOGGER_NAME);
+ protected static final int HA_ADDRESS_MIN_LENGTH = 6;
+
+ protected final BrokerConfig brokerConfig;
private final NettyServerConfig nettyServerConfig;
private final NettyClientConfig nettyClientConfig;
- private final MessageStoreConfig messageStoreConfig;
- private final ConsumerOffsetManager consumerOffsetManager;
- private final ConsumerManager consumerManager;
- private final ConsumerFilterManager consumerFilterManager;
- private final ProducerManager producerManager;
- private final ClientHousekeepingService clientHousekeepingService;
- private final PullMessageProcessor pullMessageProcessor;
- private final PullRequestHoldService pullRequestHoldService;
- private final MessageArrivingListener messageArrivingListener;
- private final Broker2Client broker2Client;
- private final SubscriptionGroupManager subscriptionGroupManager;
- private final ConsumerIdsChangeListener consumerIdsChangeListener;
+ protected final MessageStoreConfig messageStoreConfig;
+ protected final ConsumerOffsetManager consumerOffsetManager;
+ protected final BroadcastOffsetManager broadcastOffsetManager;
+ protected final ConsumerManager consumerManager;
+ protected final ConsumerFilterManager consumerFilterManager;
+ protected final ConsumerOrderInfoManager consumerOrderInfoManager;
+ protected final PopInflightMessageCounter popInflightMessageCounter;
+ protected final ProducerManager producerManager;
+ protected final ScheduleMessageService scheduleMessageService;
+ protected final ClientHousekeepingService clientHousekeepingService;
+ protected final PullMessageProcessor pullMessageProcessor;
+ protected final PeekMessageProcessor peekMessageProcessor;
+ protected final PopMessageProcessor popMessageProcessor;
+ protected final AckMessageProcessor ackMessageProcessor;
+ protected final ChangeInvisibleTimeProcessor changeInvisibleTimeProcessor;
+ protected final NotificationProcessor notificationProcessor;
+ protected final PollingInfoProcessor pollingInfoProcessor;
+ protected final QueryAssignmentProcessor queryAssignmentProcessor;
+ protected final ClientManageProcessor clientManageProcessor;
+ protected final SendMessageProcessor sendMessageProcessor;
+ protected final ReplyMessageProcessor replyMessageProcessor;
+ protected final PullRequestHoldService pullRequestHoldService;
+ protected final MessageArrivingListener messageArrivingListener;
+ protected final Broker2Client broker2Client;
+ protected final ConsumerIdsChangeListener consumerIdsChangeListener;
+ protected final EndTransactionProcessor endTransactionProcessor;
private final RebalanceLockManager rebalanceLockManager = new RebalanceLockManager();
- private final BrokerOuterAPI brokerOuterAPI;
- private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
- "BrokerControllerScheduledThread"));
- private final SlaveSynchronize slaveSynchronize;
- private final BlockingQueue sendThreadPoolQueue;
- private final BlockingQueue putThreadPoolQueue;
- private final BlockingQueue pullThreadPoolQueue;
- private final BlockingQueue replyThreadPoolQueue;
- private final BlockingQueue queryThreadPoolQueue;
- private final BlockingQueue clientManagerThreadPoolQueue;
- private final BlockingQueue