Compare commits

...

8 Commits

Author SHA1 Message Date
RongtongJin
a013c8fad1 [maven-release-plugin] prepare release rocketmq-dashboard-2.1.0 2025-08-01 16:47:58 +08:00
RongtongJin
cd262da8b1 Update Notice to 2025. 2025-07-31 11:34:24 +08:00
Crazylychee
37dbd7f327 [ISSUE #348] Fix Some interaction issues with the consumer interface (#349) 2025-07-31 11:24:42 +08:00
strangelookingnerd
79556420f5 Use provided scope for lombok (#350) 2025-07-31 11:24:01 +08:00
Crazylychee
9cb185afc1 [ISSUE #344] fix maven package display errors and npm i failed (#345)
* [ISSUE #344] fix maven package display errors and npm i failed

* fix
2025-07-16 19:21:39 +08:00
Crazylychee
f60103af9b [ISSUE #346] fix request error when logging in using the configuration file (#347) 2025-07-16 19:21:25 +08:00
TianMing2018
9c2a069976 #docs fix wrong port (#343) 2025-07-16 18:54:15 +08:00
Crazylychee
8cc7d6a727 [ISSUE #341] Add url parameter transcoding (#342)
* [Enhancement] ACL can add rules in clusters and fix ISSUE #297

* rollback the yml change

* [ISSUE #341] Add url parameter transcoding

* [ISSUE #344] fix maven package display errors and npm i failed
2025-07-16 18:53:54 +08:00
14 changed files with 80 additions and 84 deletions

2
NOTICE
View File

@@ -1,5 +1,5 @@
Apache RocketMQ
Copyright 2016-2022 The Apache Software Foundation
Copyright 2016-2025 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).

View File

@@ -9,7 +9,7 @@ In the project directory, you can run:
### `npm run start`
Runs the app in the development mode.\
Open [http://localhost:3003](http://localhost:3000) to view it in your browser.
Open [http://localhost:3003](http://localhost:3003) to view it in your browser.
The page will reload when you make changes.\
You may also see any lint errors in the console.

View File

@@ -19,11 +19,11 @@
"echarts": "^5.6.0",
"framer-motion": "^12.16.0",
"http-proxy-middleware": "^3.0.5",
"i18next": "^25.1.3",
"i18next": "^23.2.3",
"moment": "^2.30.1",
"react": "^19.1.0",
"react-dom": "^19.1.0",
"react-i18next": "^15.5.1",
"react-i18next": "14.1.3",
"react-redux": "^9.2.0",
"react-router-dom": "^7.6.0",
"react-scripts": "5.0.1",
@@ -8918,9 +8918,9 @@
}
},
"node_modules/i18next": {
"version": "25.1.3",
"resolved": "https://registry.npmmirror.com/i18next/-/i18next-25.1.3.tgz",
"integrity": "sha512-VY1iKox3YWPRTNMHFdgk5TV+Jq6rNKexLCLpPmP5oXXJ5Kl7yDBi3ycZ5fyEKZ1tNBW5gOqD4WV0XqE7rl3JUg==",
"version": "23.16.8",
"resolved": "https://registry.npmmirror.com/i18next/-/i18next-23.16.8.tgz",
"integrity": "sha512-06r/TitrM88Mg5FdUXAKL96dJMzgqLE5dv3ryBAra4KCwD9mJ4ndOTS95ZuymIGoE+2hzfdaMak2X11/es7ZWg==",
"funding": [
{
"type": "individual",
@@ -8936,15 +8936,7 @@
}
],
"dependencies": {
"@babel/runtime": "^7.27.1"
},
"peerDependencies": {
"typescript": "^5"
},
"peerDependenciesMeta": {
"typescript": {
"optional": true
}
"@babel/runtime": "^7.23.2"
}
},
"node_modules/iconv-lite": {
@@ -14000,17 +13992,16 @@
"integrity": "sha512-SN/U6Ytxf1QGkw/9ve5Y+NxBbZM6Ht95tuXNMKs8EJyFa/Vy/+Co3stop3KBHARfn/giv+Lj1uUnTfOJ3moFEQ=="
},
"node_modules/react-i18next": {
"version": "15.5.1",
"resolved": "https://registry.npmmirror.com/react-i18next/-/react-i18next-15.5.1.tgz",
"integrity": "sha512-C8RZ7N7H0L+flitiX6ASjq9p5puVJU1Z8VyL3OgM/QOMRf40BMZX+5TkpxzZVcTmOLPX5zlti4InEX5pFyiVeA==",
"version": "14.1.3",
"resolved": "https://registry.npmmirror.com/react-i18next/-/react-i18next-14.1.3.tgz",
"integrity": "sha512-wZnpfunU6UIAiJ+bxwOiTmBOAaB14ha97MjOEnLGac2RJ+h/maIYXZuTHlmyqQVX1UVHmU1YDTQ5vxLmwfXTjw==",
"dependencies": {
"@babel/runtime": "^7.25.0",
"@babel/runtime": "^7.23.9",
"html-parse-stringify": "^3.0.1"
},
"peerDependencies": {
"i18next": ">= 23.2.3",
"react": ">= 16.8.0",
"typescript": "^5"
"react": ">= 16.8.0"
},
"peerDependenciesMeta": {
"react-dom": {
@@ -14018,9 +14009,6 @@
},
"react-native": {
"optional": true
},
"typescript": {
"optional": true
}
}
},

View File

@@ -14,11 +14,11 @@
"echarts": "^5.6.0",
"framer-motion": "^12.16.0",
"http-proxy-middleware": "^3.0.5",
"i18next": "^25.1.3",
"i18next": "^23.2.3",
"moment": "^2.30.1",
"react": "^19.1.0",
"react-dom": "^19.1.0",
"react-i18next": "^15.5.1",
"react-i18next": "14.1.3",
"react-redux": "^9.2.0",
"react-router-dom": "^7.6.0",
"react-scripts": "5.0.1",

View File

@@ -316,7 +316,7 @@ const remoteApi = {
*/
queryMessageByTopicAndKey: async (topic, key) => {
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/message/queryMessageByTopicAndKey.query?topic=${topic}&key=${key}`));
const response = await remoteApi._fetch(remoteApi.buildUrl(`/message/queryMessageByTopicAndKey.query?topic=${encodeURIComponent(topic)}&key=${key}`));
const data = await response.json();
return data;
} catch (error) {
@@ -355,6 +355,7 @@ const remoteApi = {
*/
resendMessageDirectly: async (msgId, consumerGroup, topic) => {
topic = encodeURIComponent(topic)
consumerGroup = encodeURIComponent(consumerGroup)
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/message/consumeMessageDirectly.do?msgId=${msgId}&consumerGroup=${consumerGroup}&topic=${topic}`), {
method: 'POST',
@@ -392,6 +393,7 @@ const remoteApi = {
},
refreshConsumerGroup: async (consumerGroup) => {
consumerGroup = encodeURIComponent(consumerGroup)
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/consumer/group.refresh?consumerGroup=${consumerGroup}`));
const data = await response.json();
@@ -443,6 +445,7 @@ const remoteApi = {
},
fetchBrokerNameList: async (consumerGroup) => {
consumerGroup = encodeURIComponent(consumerGroup)
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/consumer/fetchBrokerNameList.query?consumerGroup=${consumerGroup}`));
const data = await response.json();
@@ -454,6 +457,7 @@ const remoteApi = {
},
deleteConsumerGroup: async (groupName, brokerNameList) => {
groupName = encodeURIComponent(groupName)
try {
const response = await remoteApi._fetch(remoteApi.buildUrl("/consumer/deleteSubGroup.do"), {
method: 'POST',
@@ -471,6 +475,7 @@ const remoteApi = {
},
queryConsumerConfig: async (consumerGroup) => {
consumerGroup = encodeURIComponent(consumerGroup)
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/consumer/examineSubscriptionGroupConfig.query?consumerGroup=${consumerGroup}`));
const data = await response.json();
@@ -499,6 +504,7 @@ const remoteApi = {
},
queryTopicByConsumer: async (consumerGroup, address) => {
consumerGroup = encodeURIComponent(consumerGroup)
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/consumer/queryTopicByConsumer.query?consumerGroup=${consumerGroup}&address=${address}`));
const data = await response.json();
@@ -510,6 +516,7 @@ const remoteApi = {
},
queryConsumerConnection: async (consumerGroup, address) => {
consumerGroup = encodeURIComponent(consumerGroup)
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/consumer/consumerConnection.query?consumerGroup=${consumerGroup}&address=${address}`));
const data = await response.json();
@@ -521,6 +528,7 @@ const remoteApi = {
},
queryConsumerRunningInfo: async (consumerGroup, clientId, jstack = false) => {
consumerGroup = encodeURIComponent(consumerGroup)
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/consumer/consumerRunningInfo.query?consumerGroup=${consumerGroup}&clientId=${clientId}&jstack=${jstack}`));
const data = await response.json();
@@ -559,7 +567,7 @@ const remoteApi = {
getTopicStats: async (topic) => {
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/stats.query?topic=${topic}`));
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/stats.query?topic=${encodeURIComponent(topic)}`));
return await response.json();
} catch (error) {
console.error("Error fetching topic stats:", error);
@@ -569,7 +577,7 @@ const remoteApi = {
getTopicRoute: async (topic) => {
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/route.query?topic=${topic}`));
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/route.query?topic=${encodeURIComponent(topic)}`));
return await response.json();
} catch (error) {
console.error("Error fetching topic route:", error);
@@ -579,7 +587,7 @@ const remoteApi = {
getTopicConsumers: async (topic) => {
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/queryConsumerByTopic.query?topic=${topic}`));
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/queryConsumerByTopic.query?topic=${encodeURIComponent(topic)}`));
return await response.json();
} catch (error) {
console.error("Error fetching topic consumers:", error);
@@ -589,7 +597,7 @@ const remoteApi = {
getTopicConsumerGroups: async (topic) => {
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/queryTopicConsumerInfo.query?topic=${topic}`));
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/queryTopicConsumerInfo.query?topic=${encodeURIComponent(topic)}`));
return await response.json();
} catch (error) {
console.error("Error fetching consumer groups:", error);
@@ -599,7 +607,7 @@ const remoteApi = {
getTopicConfig: async (topic) => {
try {
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/examineTopicConfig.query?topic=${topic}`));
const response = await remoteApi._fetch(remoteApi.buildUrl(`/topic/examineTopicConfig.query?topic=${encodeURIComponent(topic)}`));
return await response.json();
} catch (error) {
console.error("Error fetching topic config:", error);

View File

@@ -21,7 +21,7 @@ import {remoteApi} from '../../api/remoteApi/remoteApi';
import {useLanguage} from '../../i18n/LanguageContext';
const ClientInfoModal = ({visible, group, address, onCancel}) => {
const ClientInfoModal = ({visible, group, address, onCancel, messageApi}) => {
const {t} = useLanguage();
const [loading, setLoading] = useState(false);
const [connectionData, setConnectionData] = useState(null);
@@ -34,7 +34,11 @@ const ClientInfoModal = ({visible, group, address, onCancel}) => {
try {
const connResponse = await remoteApi.queryConsumerConnection(group, address);
if (connResponse.status === 0) setConnectionData(connResponse.data);
if (connResponse.status === 0) {
setConnectionData(connResponse.data);
}else{
messageApi.error(connResponse.errMsg);
}
} finally {
setLoading(false);
}

View File

@@ -20,7 +20,7 @@ import {Modal, Spin, Table} from 'antd';
import {remoteApi} from '../../api/remoteApi/remoteApi';
import {useLanguage} from '../../i18n/LanguageContext';
const ConsumerDetailModal = ({visible, group, address, onCancel}) => {
const ConsumerDetailModal = ({visible, group, address, onCancel ,messageApi}) => {
const {t} = useLanguage();
const [loading, setLoading] = useState(false);
const [details, setDetails] = useState([]);
@@ -34,6 +34,10 @@ const ConsumerDetailModal = ({visible, group, address, onCancel}) => {
const response = await remoteApi.queryTopicByConsumer(group, address);
if (response.status === 0) {
setDetails(response.data);
}else {
// Handle error case
messageApi.error(response.errMsg);
setDetails([]);
}
} finally {
setLoading(false);

View File

@@ -18,15 +18,12 @@
import React, {useEffect, useState} from 'react';
import {Button, Checkbox, Modal, notification, Spin} from 'antd';
import {remoteApi} from '../../api/remoteApi/remoteApi';
import {useLanguage} from '../../i18n/LanguageContext';
const DeleteConsumerModal = ({visible, group, onCancel, onSuccess}) => {
const {t} = useLanguage();
const DeleteConsumerModal = ({visible, group, onCancel, onSuccess, t}) => {
const [brokerList, setBrokerList] = useState([]);
const [selectedBrokers, setSelectedBrokers] = useState([]);
const [loading, setLoading] = useState(false);
// 获取Broker列表
useEffect(() => {
const fetchBrokers = async () => {
if (!visible) return;
@@ -45,7 +42,6 @@ const DeleteConsumerModal = ({visible, group, onCancel, onSuccess}) => {
fetchBrokers();
}, [visible, group]);
// 处理删除提交
const handleDelete = async () => {
if (selectedBrokers.length === 0) {
notification.warning({message: t.PLEASE_SELECT_BROKER});

View File

@@ -592,9 +592,10 @@ export const translations = {
"EXPRESSION_TYPE": "Expression Type",
"SUB_VERSION": "Sub Version",
"CODE_SET": "Code Set",
"TAGS_SET": "Tags Set"
"TAGS_SET": "Tags Set",
"DELETE_CONSUMER_GROUP": "Delete Consumer Group",
"SELECT_DELETE_BROKERS": "Please select brokers to delete consumer group",
"CONFIRM_DELETE": "Confirm Delete",
}
};

View File

@@ -288,7 +288,6 @@ const ConsumerGroupList = () => {
setShowConfig(true);
};
// 修改操作按钮的点击处理函数
const handleClient = (group, address) => {
setSelectedGroup(group);
setSelectedAddress(address);
@@ -550,12 +549,12 @@ const ConsumerGroupList = () => {
/>
</Spin>
{/* 模态框组件保持不变 */}
<ClientInfoModal
visible={showClientInfo}
group={selectedGroup}
address={selectedAddress}
onCancel={() => setShowClientInfo(false)}
messageApi={messageApi}
/>
<ConsumerDetailModal
@@ -563,6 +562,7 @@ const ConsumerGroupList = () => {
group={selectedGroup}
address={selectedAddress}
onCancel={() => setShowConsumeDetail(false)}
messageApi={messageApi}
/>
<ConsumerConfigModal
@@ -579,6 +579,7 @@ const ConsumerGroupList = () => {
group={selectedGroup}
onCancel={() => setShowDeleteModal(false)}
onSuccess={loadConsumerGroups}
t={t}
/>
</div>
</>

12
pom.xml
View File

@@ -16,8 +16,7 @@
~ limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache</groupId>
@@ -29,14 +28,14 @@
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-dashboard</artifactId>
<packaging>jar</packaging>
<version>2.0.1-SNAPSHOT</version>
<version>2.1.0</version>
<name>rocketmq-dashboard</name>
<scm>
<url>git@github.com:apache/rocketmq-dashboard.git</url>
<connection>scm:git:git@github.com:apache/rocketmq-dashboard.git</connection>
<developerConnection>scm:git:git@github.com:apache/rocketmq-dashboard.git</developerConnection>
<tag>1.0.0</tag>
<tag>rocketmq-dashboard-2.1.0</tag>
</scm>
<mailingLists>
@@ -250,6 +249,7 @@
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>${lombok.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
@@ -448,7 +448,7 @@
<goal>npm</goal>
</goals>
<configuration>
<arguments>install --legacy-peer-deps</arguments>
<arguments>install</arguments>
</configuration>
</execution>
<execution>
@@ -470,7 +470,7 @@
<configuration>
<target>
<copy todir="${project.build.directory}/classes/public">
<fileset dir="${project.basedir}/frontend-new/build"/>
<fileset dir="${project.basedir}/frontend-new/build" />
</copy>
</target>
</configuration>

View File

@@ -80,7 +80,7 @@ public class MQAdminAspect {
String methodName = joinPoint.getSignature().getName();
try {
if (isPoolConfigIsolatedByUser(rmqConfigure.isLoginRequired(), methodName)) {
if (isPoolConfigIsolatedByUser(rmqConfigure.isLoginRequired(), rmqConfigure.getAuthMode(), methodName)) {
currentUserInfo = (UserInfo) UserInfoContext.get(WebUtil.USER_NAME);
// 2. Borrow the user-specific MQAdminExt instance.
// currentUser.getName() is assumed to be the AccessKey, and currentUser.getPassword() is SecretKey.
@@ -123,8 +123,8 @@ public class MQAdminAspect {
}
}
private boolean isPoolConfigIsolatedByUser(boolean loginRequired, String methodName) {
if (!loginRequired) {
private boolean isPoolConfigIsolatedByUser(boolean loginRequired, String authMode, String methodName) {
if (!loginRequired || authMode.equals("file")) {
return false;
} else {
return !METHODS_TO_CHECK.contains(methodName);

View File

@@ -283,6 +283,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
@Override
public List<TopicConsumerInfo> queryConsumeStatsListByGroupName(String groupName, String address) {
groupName = getConsumerGroup(groupName);
List<ConsumeStats> consumeStatses = new ArrayList<>();
String topic = null;
try {
@@ -295,9 +296,10 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
throw new RuntimeException(e);
}
List<TopicConsumerInfo> res = new ArrayList<>();
String finalGroupName = groupName;
consumeStatses.forEach(consumeStats -> {
if (consumeStats != null && consumeStats.getOffsetTable() != null && !consumeStats.getOffsetTable().isEmpty()) {
res.addAll(toTopicConsumerInfoList(topic, consumeStats, groupName));
res.addAll(toTopicConsumerInfoList(topic, consumeStats, finalGroupName));
}
});
return res;
@@ -305,6 +307,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
@Override
public List<TopicConsumerInfo> queryConsumeStatsList(final String topic, String groupName) {
groupName = getConsumerGroup(groupName);
ConsumeStats consumeStats = null;
try {
consumeStats = mqAdminExt.examineConsumeStats(groupName, topic);
@@ -316,6 +319,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
private List<TopicConsumerInfo> toTopicConsumerInfoList(String topic, ConsumeStats consumeStats, String groupName) {
groupName = getConsumerGroup(groupName);
List<MessageQueue> mqList = Lists.newArrayList(Iterables.filter(consumeStats.getOffsetTable().keySet(), new Predicate<MessageQueue>() {
@Override
public boolean apply(MessageQueue o) {
@@ -339,6 +343,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
private Map<MessageQueue, String> getClientConnection(String groupName) {
groupName = getConsumerGroup(groupName);
Map<MessageQueue, String> results = Maps.newHashMap();
try {
ConsumerConnection consumerConnection = mqAdminExt.examineConsumerConnectionInfo(groupName);
@@ -417,7 +422,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
@Override
public List<ConsumerConfigInfo> examineSubscriptionGroupConfig(String group) {
public List<ConsumerConfigInfo> examineSubscriptionGroupConfig(String consumerGroup) {
consumerGroup = getConsumerGroup(consumerGroup);
List<ConsumerConfigInfo> consumerConfigInfoList = Lists.newArrayList();
try {
ClusterInfo clusterInfo = clusterInfoService.get();
@@ -425,9 +431,9 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
String brokerAddress = clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr();
SubscriptionGroupConfig subscriptionGroupConfig = null;
try {
subscriptionGroupConfig = mqAdminExt.examineSubscriptionGroupConfig(brokerAddress, group);
subscriptionGroupConfig = mqAdminExt.examineSubscriptionGroupConfig(brokerAddress, consumerGroup);
} catch (Exception e) {
logger.warn("op=examineSubscriptionGroupConfig_error brokerName={} group={}", brokerName, group);
logger.warn("op=examineSubscriptionGroupConfig_error brokerName={} group={}", brokerName, consumerGroup);
}
if (subscriptionGroupConfig == null) {
continue;
@@ -480,6 +486,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
@Override
public boolean createAndUpdateSubscriptionGroupConfig(ConsumerConfigInfo consumerConfigInfo) {
consumerConfigInfo.getSubscriptionGroupConfig().setGroupName(getConsumerGroup(consumerConfigInfo.getSubscriptionGroupConfig().getGroupName()));
try {
ClusterInfo clusterInfo = clusterInfoService.get();
for (String brokerName : changeToBrokerNameSet(clusterInfo.getClusterAddrTable(),
@@ -495,6 +502,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
@Override
public Set<String> fetchBrokerNameSetBySubscriptionGroup(String group) {
group = getConsumerGroup(group);
Set<String> brokerNameSet = Sets.newHashSet();
try {
List<ConsumerConfigInfo> consumerConfigInfoList = examineSubscriptionGroupConfig(group);
@@ -511,6 +519,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
@Override
public ConsumerConnection getConsumerConnection(String consumerGroup, String address) {
consumerGroup = getConsumerGroup(consumerGroup);
try {
String[] addresses = address.split(",");
String addr = addresses[0];
@@ -523,6 +532,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
@Override
public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack) {
consumerGroup = getConsumerGroup(consumerGroup);
try {
return mqAdminExt.getConsumerRunningInfo(consumerGroup, clientId, jstack);
} catch (Exception e) {
@@ -533,7 +543,6 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
@Override
public GroupConsumeInfo refreshGroup(String address, String consumerGroup) {
if (isCacheBeingBuilt || cacheConsumeInfoList.isEmpty()) {
throw new RuntimeException("Cache is being built or empty, please try again later");
}
@@ -541,7 +550,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
for (int i = 0; i < cacheConsumeInfoList.size(); i++) {
GroupConsumeInfo groupConsumeInfo = cacheConsumeInfoList.get(i);
if (groupConsumeInfo.getGroup().equals(consumerGroup)) {
GroupConsumeInfo updatedInfo = queryGroup(consumerGroup, "");
GroupConsumeInfo updatedInfo = queryGroup(consumerGroup, address);
updatedInfo.setUpdateTime(new Date());
updatedInfo.setGroup(consumerGroup);
updatedInfo.setAddress(consumerGroupMap.get(consumerGroup));
@@ -559,4 +568,11 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
consumerGroupMap.clear();
return queryGroupList(false, address);
}
public String getConsumerGroup(String consumerGroup) {
if (consumerGroup != null && consumerGroup.startsWith("%SYS%")) {
return consumerGroup.substring(5); // Remove "%SYS%" prefix
}
return consumerGroup;
}
}

View File

@@ -21,11 +21,9 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.impl.MQAdminImpl;
import org.apache.rocketmq.client.impl.MQClientAPIImpl;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.dashboard.service.client.MQAdminExtImpl;
@@ -589,26 +587,6 @@ public class MQAdminExtImplTest {
Assert.assertNotNull(result);
}
@Test
public void testStart() {
assertNotNull(mqAdminExtImpl);
try {
mqAdminExtImpl.start();
} catch (Exception e) {
Assert.assertTrue(e instanceof IllegalStateException);
}
}
@Test
public void testShutdown() {
assertNotNull(mqAdminExtImpl);
try {
mqAdminExtImpl.shutdown();
} catch (Exception e) {
Assert.assertTrue(e instanceof IllegalStateException);
}
}
@Test
public void testQueryConsumeTimeSpan() throws Exception {
assertNotNull(mqAdminExtImpl);
@@ -659,7 +637,7 @@ public class MQAdminExtImplTest {
{
when(defaultMQAdminExt.viewBrokerStatsData(anyString(), anyString(), anyString())).thenReturn(new BrokerStatsData());
}
BrokerStatsData brokerStatsData = mqAdminExtImpl.viewBrokerStatsData(brokerAddr, BrokerStatsManager.TOPIC_PUT_NUMS, "topic_test");
BrokerStatsData brokerStatsData = mqAdminExtImpl.viewBrokerStatsData(brokerAddr, BrokerStatsManager.BROKER_ACK_NUMS, "topic_test");
Assert.assertNotNull(brokerStatsData);
}