This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new e6c2a7fe1 fix(metadata): scope persisted topic/group identity by
instance (#2366)
e6c2a7fe1 is described below
commit e6c2a7fe1a944fccdeeac64ba6bb60edaf3c5676
Author: aias00 <[email protected]>
AuthorDate: Wed Aug 19 14:56:27 2026 +0800
fix(metadata): scope persisted topic/group identity by instance (#2366)
Constraint: Existing topic/group records were keyed by cluster plus name,
so same-name resources under different Studio instances could overwrite or
delete each other.
Rejected: Keep nullable instance_id inside the new unique key | MySQL
unique indexes still allow duplicate NULL scopes, so legacy rows would remain
ambiguous.
Directive: Persist and query topic/group metadata with the (cluster_id,
instance_id, name) identity; use the empty-string legacy scope only for old
unassigned rows.
Confidence: high
Scope-risk: moderate
Tested:
JAVA_HOME=/Users/aias/Library/Java/JavaVirtualMachines/openjdk-21.0.2/Contents/Home
mvn
-Dtest=RocketMQAdminClientImplTest,RocketMQMetadataProviderTest,ApacheInstanceProviderTest,MybatisPlusInstanceRepositoryTest
test
Signed-off-by: liuhy <[email protected]>
---
.../studio/instance/InstanceRepository.java | 4 +--
.../instance/MybatisPlusInstanceRepository.java | 4 +--
.../provider/apache/ApacheInstanceProvider.java | 17 ++-------
.../studio/provider/apache/MetadataProvider.java | 4 +++
.../provider/apache/RocketMQAdminClientImpl.java | 41 +++++++++++++---------
.../provider/apache/RocketMQMetadataProvider.java | 14 ++++++--
server/src/main/resources/db/schema.sql | 8 ++---
.../MybatisPlusInstanceRepositoryTest.java | 4 +--
.../apache/ApacheInstanceProviderTest.java | 13 +++++--
.../apache/RocketMQAdminClientImplTest.java | 13 ++++---
.../apache/RocketMQMetadataProviderTest.java | 34 ++++++++++++++++++
11 files changed, 107 insertions(+), 49 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
index a4e8ce716..89173ce3a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
@@ -60,7 +60,7 @@ public interface InstanceRepository {
boolean existsByCredentialId(Long credentialId);
- long countTopicsByInstance(Long instanceId);
+ long countTopicsByInstance(String instanceId);
- long countGroupsByInstance(Long instanceId);
+ long countGroupsByInstance(String instanceId);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepository.java
index 0d49534ae..422e8643a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepository.java
@@ -152,13 +152,13 @@ public class MybatisPlusInstanceRepository implements
InstanceRepository {
}
@Override
- public long countTopicsByInstance(Long instanceId) {
+ public long countTopicsByInstance(String instanceId) {
return topicMapper.selectCount(
new QueryWrapper<RmqTopic>().eq("instance_id", instanceId));
}
@Override
- public long countGroupsByInstance(Long instanceId) {
+ public long countGroupsByInstance(String instanceId) {
return groupMapper.selectCount(
new QueryWrapper<RmqGroup>().eq("instance_id", instanceId));
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
index 6c4147ecb..27b1fc909 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
@@ -34,7 +34,6 @@ import org.springframework.stereotype.Component;
import lombok.RequiredArgsConstructor;
import java.util.List;
-import java.util.Objects;
import java.util.Set;
/**
@@ -68,24 +67,21 @@ public class ApacheInstanceProvider implements
InstanceProvider {
@Override
public int countTopics(String instanceId) {
- // The instance_id foreign key is numeric; resolve the external
identifier first.
return instanceRepository.findByIdentifier(instanceId)
- .map(instance -> (int)
instanceRepository.countTopicsByInstance(instance.getId()))
+ .map(instance -> (int)
instanceRepository.countTopicsByInstance(instance.getName()))
.orElse(0);
}
@Override
public int countGroups(String instanceId) {
return instanceRepository.findByIdentifier(instanceId)
- .map(instance -> (int)
instanceRepository.countGroupsByInstance(instance.getId()))
+ .map(instance -> (int)
instanceRepository.countGroupsByInstance(instance.getName()))
.orElse(0);
}
@Override
public List<TopicVO> listTopics(String instanceId, String type, String
search) {
- return metadataProvider.listTopics(null, type, search).stream()
- .filter(topic -> matchesInstance(topic.getInstanceId(),
instanceId))
- .toList();
+ return metadataProvider.listTopics(instanceId, null, type, search);
}
@Override
@@ -159,11 +155,4 @@ public class ApacheInstanceProvider implements
InstanceProvider {
public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic) {
return messageProvider.getMessageTrace(instanceId, msgId, topic);
}
-
- private boolean matchesInstance(String topicInstanceId, String instanceId)
{
- if (instanceId == null || instanceId.isBlank()) {
- return true;
- }
- return Objects.equals(topicInstanceId, instanceId);
- }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
index 181267c8f..7e16d005e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
@@ -31,6 +31,10 @@ import java.util.List;
public interface MetadataProvider {
List<TopicVO> listTopics(String clusterId, String type, String search);
+ default List<TopicVO> listTopics(String instanceId, String clusterId,
String type, String search) {
+ return listTopics(clusterId, type, search);
+ }
+
default PageResult<TopicVO> listTopicsPage(String clusterId, String type,
String search,
int page, int pageSize) {
List<TopicVO> topics = listTopics(clusterId, type, search);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index 2eefa9c7a..a16e8b849 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -70,6 +70,7 @@ public class RocketMQAdminClientImpl implements AdminClient {
private static final String MESSAGE_SENDER_GROUP_PREFIX =
"studio-msg-sender";
private static final int MAX_MESSAGE_SIZE = 4 * 1024 * 1024; // 4 MB
default broker limit
+ private static final String LEGACY_METADATA_SCOPE = "";
private final MqAdminExtFactory adminFactory;
private final RocketMQProperties properties;
@@ -142,11 +143,12 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return executeForInstance(topic.getInstanceId(), admin -> {
try {
String clusterName = getClusterName(admin);
- // Match on the (cluster_id, name) key: the same topic name
can exist in several
- // clusters, and a name-only lookup would blow up with
TooManyResultsException.
+ // Match on the instance-scoped identity: the same topic name
can exist in several
+ // clusters or Studio instances, and a broader lookup would
corrupt metadata.
RmqTopic existing = topicMapper.selectOne(
new LambdaQueryWrapper<RmqTopic>()
.eq(RmqTopic::getClusterId, clusterName)
+ .eq(RmqTopic::getInstanceId,
metadataScope(topic.getInstanceId()))
.eq(RmqTopic::getName, topicName));
TopicPerm effectivePerm = topic.getPerm() != null
? topic.getPerm()
@@ -169,20 +171,20 @@ public class RocketMQAdminClientImpl implements
AdminClient {
// Persist to DB. Re-creating a topic that already has a
record (for example when
// rebuilding a broker route from the console) must update it
instead of failing on
- // the unique (cluster_id, name) key.
+ // the instance-scoped unique key.
RmqTopic entity = topicMapper.selectOne(new
LambdaQueryWrapper<RmqTopic>()
.eq(RmqTopic::getClusterId, clusterName)
+ .eq(RmqTopic::getInstanceId,
metadataScope(topic.getInstanceId()))
.eq(RmqTopic::getName, topicName));
boolean isNew = entity == null;
if (isNew) {
entity = new RmqTopic();
entity.setName(topicName);
entity.setClusterId(clusterName);
+ entity.setInstanceId(metadataScope(topic.getInstanceId()));
entity.setGmtCreate(LocalDateTime.now());
}
- if (topic.getInstanceId() != null) {
- entity.setInstanceId(topic.getInstanceId());
- }
+ entity.setInstanceId(metadataScope(topic.getInstanceId()));
if (topic.getType() != null) {
entity.setTopicType(topic.getType().name());
} else if (isNew) {
@@ -225,12 +227,13 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return executeForInstance(topic.getInstanceId(), admin -> {
try {
- // Match on the (cluster_id, name) key to avoid ambiguity when
several clusters share
- // the same topic name (a name-only lookup would throw
TooManyResultsException).
+ // Match on the instance-scoped identity so same-name
resources under different
+ // Studio instances do not overwrite each other.
String clusterName = getClusterName(admin);
RmqTopic existing = topicMapper.selectOne(
new LambdaQueryWrapper<RmqTopic>()
.eq(RmqTopic::getClusterId, clusterName)
+ .eq(RmqTopic::getInstanceId,
metadataScope(topic.getInstanceId()))
.eq(RmqTopic::getName, topicName));
// Preserve the existing queue counts when the update request
does not change them,
// matching the perm semantics below; defaulting to 8 would
silently resize the
@@ -329,6 +332,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
// Topic names may be shared by several clusters managed by
this Studio instance.
topicMapper.delete(new LambdaQueryWrapper<RmqTopic>()
.eq(RmqTopic::getClusterId, clusterName)
+ .eq(RmqTopic::getInstanceId, metadataScope(instanceId))
.eq(RmqTopic::getName, name));
recordAudit("DELETE_TOPIC", name, "", "SUCCESS");
@@ -437,21 +441,21 @@ public class RocketMQAdminClientImpl implements
AdminClient {
admin.createAndUpdateSubscriptionGroupConfig(addr, config);
}
- // Persist to DB, upserting so re-creating an existing group does
not violate the
- // unique (cluster_id, name) key.
+ // Persist to DB, upserting so re-creating an existing group stays
scoped to the
+ // selected Studio instance.
RmqGroup entity = groupMapper.selectOne(new
LambdaQueryWrapper<RmqGroup>()
.eq(RmqGroup::getClusterId, groupClusterName)
+ .eq(RmqGroup::getInstanceId,
metadataScope(group.getInstanceId()))
.eq(RmqGroup::getName, groupName));
boolean isNewGroup = entity == null;
if (isNewGroup) {
entity = new RmqGroup();
entity.setName(groupName);
entity.setClusterId(groupClusterName);
+ entity.setInstanceId(metadataScope(group.getInstanceId()));
entity.setGmtCreate(LocalDateTime.now());
}
- if (group.getInstanceId() != null) {
- entity.setInstanceId(group.getInstanceId());
- }
+ entity.setInstanceId(metadataScope(group.getInstanceId()));
entity.setConsumeType(group.getConsumeType() != null ?
group.getConsumeType().name() : "CLUSTERING");
entity.setMessageModel(group.getSubscriptionMode() != null ?
group.getSubscriptionMode().name() : "Push");
entity.setMaxRetry(config.getRetryMaxTimes());
@@ -481,18 +485,18 @@ public class RocketMQAdminClientImpl implements
AdminClient {
public void deleteConsumerGroup(String instanceId, String name) {
if (StringUtils.hasText(instanceId)) {
runtimeAdminClientResolver.execute(instanceId, admin -> {
- doDeleteConsumerGroup(admin, name);
+ doDeleteConsumerGroup(instanceId, admin, name);
return null;
});
return;
}
adminFactory.execute(namesrvAddr(), null, admin -> {
- doDeleteConsumerGroup(admin, name);
+ doDeleteConsumerGroup(null, admin, name);
return null;
});
}
- private void doDeleteConsumerGroup(MQAdminExt admin, String name) {
+ private void doDeleteConsumerGroup(String instanceId, MQAdminExt admin,
String name) {
try {
String clusterName = getClusterName(admin);
Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin,
clusterName);
@@ -504,6 +508,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
// Consumer group names may be shared by several clusters managed
by this Studio instance.
groupMapper.delete(new LambdaQueryWrapper<RmqGroup>()
.eq(RmqGroup::getClusterId, clusterName)
+ .eq(RmqGroup::getInstanceId, metadataScope(instanceId))
.eq(RmqGroup::getName, name));
recordAudit("DELETE_GROUP", name, "", "SUCCESS");
@@ -613,6 +618,10 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return adminFactory.execute(namesrvAddr(), null, action);
}
+ private String metadataScope(String instanceId) {
+ return StringUtils.hasText(instanceId) ? instanceId.trim() :
LEGACY_METADATA_SCOPE;
+ }
+
private Set<String> getAllMasterBrokerAddrs(MQAdminExt admin) throws
Exception {
Set<String> addrs = new HashSet<>();
ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index df46661ed..48bfdc935 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -116,7 +116,13 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
@Override
public List<TopicVO> listTopics(String clusterId, String type, String
search) {
+ return listTopics(null, clusterId, type, search);
+ }
+
+ @Override
+ public List<TopicVO> listTopics(String instanceId, String clusterId,
String type, String search) {
LambdaQueryWrapper<RmqTopic> query = new LambdaQueryWrapper<RmqTopic>()
+ .eq(instanceId != null, RmqTopic::getInstanceId,
normalizeMetadataScope(instanceId))
.eq(StringUtils.hasText(clusterId), RmqTopic::getClusterId,
clusterId)
.eq(StringUtils.hasText(type), RmqTopic::getTopicType, type)
.like(StringUtils.hasText(search), RmqTopic::getName, search)
@@ -136,7 +142,7 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
public PageResult<TopicVO> listTopicsPage(String instanceId, String
clusterId, String type,
String search, int page, int pageSize) {
LambdaQueryWrapper<RmqTopic> query = new LambdaQueryWrapper<RmqTopic>()
- .eq(StringUtils.hasText(instanceId), RmqTopic::getInstanceId,
instanceId)
+ .eq(instanceId != null, RmqTopic::getInstanceId,
normalizeMetadataScope(instanceId))
.eq(StringUtils.hasText(clusterId), RmqTopic::getClusterId,
clusterId)
.eq(StringUtils.hasText(type), RmqTopic::getTopicType, type)
.like(StringUtils.hasText(search), RmqTopic::getName, search)
@@ -195,7 +201,7 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
@Override
public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String
clusterId, String search) {
LambdaQueryWrapper<RmqGroup> query = new LambdaQueryWrapper<RmqGroup>()
- .eq(StringUtils.hasText(instanceId), RmqGroup::getInstanceId,
instanceId)
+ .eq(instanceId != null, RmqGroup::getInstanceId,
normalizeMetadataScope(instanceId))
.eq(StringUtils.hasText(clusterId), RmqGroup::getClusterId,
clusterId)
.like(StringUtils.hasText(search), RmqGroup::getName, search)
.orderByAsc(RmqGroup::getName);
@@ -225,6 +231,10 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
return result;
}
+ private String normalizeMetadataScope(String instanceId) {
+ return StringUtils.hasText(instanceId) ? instanceId.trim() : "";
+ }
+
private ConsumeType parseConsumeType(String messageModel) {
if (!StringUtils.hasText(messageModel)) {
return ConsumeType.CLUSTERING;
diff --git a/server/src/main/resources/db/schema.sql
b/server/src/main/resources/db/schema.sql
index 9132df72a..f20f859d6 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -80,7 +80,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_topic (
`gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT '修改时间',
cluster_id VARCHAR(64) NOT NULL,
- instance_id VARCHAR(128) COMMENT '归属实例 ID（rmq_instance.name，全局唯一）',
+ instance_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '归属实例
ID（rmq_instance.name；空字符串表示历史未归属记录）',
name VARCHAR(255) NOT NULL,
topic_type VARCHAR(32) DEFAULT 'NORMAL',
read_queue_nums INT DEFAULT 8,
@@ -90,7 +90,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_topic (
status VARCHAR(32) DEFAULT 'ACTIVE',
created_by VARCHAR(64),
PRIMARY KEY (`id`),
- UNIQUE KEY uk_cluster_topic (cluster_id, name),
+ UNIQUE KEY uk_cluster_instance_topic (cluster_id, instance_id, name),
INDEX idx_topic_instance (instance_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
@@ -100,7 +100,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_group (
`gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT '修改时间',
cluster_id VARCHAR(64) NOT NULL,
- instance_id VARCHAR(128) COMMENT '归属实例 ID（rmq_instance.name，全局唯一）',
+ instance_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '归属实例
ID（rmq_instance.name；空字符串表示历史未归属记录）',
name VARCHAR(255) NOT NULL,
consume_type VARCHAR(32) DEFAULT 'CONCURRENTLY',
message_model VARCHAR(32) DEFAULT 'CLUSTERING',
@@ -108,7 +108,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_group (
status VARCHAR(32) DEFAULT 'ACTIVE',
created_by VARCHAR(64),
PRIMARY KEY (`id`),
- UNIQUE KEY uk_cluster_group (cluster_id, name),
+ UNIQUE KEY uk_cluster_instance_group (cluster_id, instance_id, name),
INDEX idx_group_instance (instance_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepositoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepositoryTest.java
index 727689dc2..ab559c1ee 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepositoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepositoryTest.java
@@ -175,7 +175,7 @@ class MybatisPlusInstanceRepositoryTest {
void countTopicsByInstanceShouldDelegateToTopicMapperTest() {
when(topicMapper.selectCount(any(QueryWrapper.class))).thenReturn(5L);
- assertThat(repository.countTopicsByInstance(2L)).isEqualTo(5L);
+
assertThat(repository.countTopicsByInstance("instance-proxy-1")).isEqualTo(5L);
verify(topicMapper).selectCount(any(QueryWrapper.class));
}
@@ -183,7 +183,7 @@ class MybatisPlusInstanceRepositoryTest {
void countGroupsByInstanceShouldDelegateToGroupMapperTest() {
when(groupMapper.selectCount(any(QueryWrapper.class))).thenReturn(2L);
- assertThat(repository.countGroupsByInstance(2L)).isEqualTo(2L);
+
assertThat(repository.countGroupsByInstance("instance-proxy-1")).isEqualTo(2L);
verify(groupMapper).selectCount(any(QueryWrapper.class));
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
index b99657915..ed63f46fb 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
@@ -72,7 +72,7 @@ class ApacheInstanceProviderTest {
InstanceVO instance = InstanceVO.builder().name("inst-1").build();
instance.setId(1L);
when(instanceRepository.findByIdentifier("inst-1")).thenReturn(Optional.of(instance));
- when(instanceRepository.countTopicsByInstance(1L)).thenReturn(3L);
+
when(instanceRepository.countTopicsByInstance("inst-1")).thenReturn(3L);
assertThat(provider.countTopics("inst-1")).isEqualTo(3);
}
@@ -82,11 +82,20 @@ class ApacheInstanceProviderTest {
InstanceVO instance = InstanceVO.builder().name("inst-1").build();
instance.setId(1L);
when(instanceRepository.findByIdentifier("inst-1")).thenReturn(Optional.of(instance));
- when(instanceRepository.countGroupsByInstance(1L)).thenReturn(2L);
+
when(instanceRepository.countGroupsByInstance("inst-1")).thenReturn(2L);
assertThat(provider.countGroups("inst-1")).isEqualTo(2);
}
+ @Test
+ void listTopicsShouldPassTheSelectedInstanceToMetadataProvider() {
+ when(metadataProvider.listTopics("inst-1", null, "FIFO",
"orders")).thenReturn(java.util.List.of());
+
+ assertThat(provider.listTopics("inst-1", "FIFO", "orders")).isEmpty();
+
+ verify(metadataProvider).listTopics("inst-1", null, "FIFO", "orders");
+ }
+
@Test
void listConsumerGroupsShouldPassTheSelectedInstanceToMetadataProvider() {
when(metadataProvider.listConsumerGroups("inst-1", null,
"orders")).thenReturn(java.util.List.of());
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index 6786c071c..ccfcf8e2c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -200,7 +200,7 @@ class RocketMQAdminClientImplTest {
}
@Test
- void createTopicScopesLookupToCluster() throws Exception {
+ void createTopicScopesLookupToClusterAndLegacyInstanceScope() throws
Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
ClusterInfo clusterInfo = new ClusterInfo();
Map<String, Set<String>> clusterAddrTable = new HashMap<>();
@@ -224,7 +224,7 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(topicMapper, times(2)).selectOne(captor.capture());
for (LambdaQueryWrapper<RmqTopic> wrapper : captor.getAllValues()) {
- assertThat(wrapper.getSqlSegment()).contains("cluster_id");
+ assertThat(wrapper.getSqlSegment()).contains("cluster_id",
"instance_id");
}
}
@@ -359,7 +359,7 @@ class RocketMQAdminClientImplTest {
}
@Test
- void topicDeleteUsesSelectedInstanceAndScopesMetadataToCluster() throws
Exception {
+ void
topicDeleteUsesSelectedInstanceAndScopesMetadataToClusterAndInstance() throws
Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
DefaultMQAdminExt selectedAdmin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
@@ -374,7 +374,7 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor<LambdaQueryWrapper<RmqTopic>> captor =
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(topicMapper).delete(captor.capture());
- assertThat(captor.getValue().getSqlSegment()).contains("cluster_id",
"name");
+ assertThat(captor.getValue().getSqlSegment()).contains("cluster_id",
"instance_id", "name");
verify(selectedAdmin).deleteTopicInBroker(Set.of("10.0.0.1:10911"),
"orders");
verify(selectedAdmin).deleteTopicInNameServer(Set.of("10.0.0.2:9876"),
"cluster-1", "orders");
}
@@ -422,6 +422,9 @@ class RocketMQAdminClientImplTest {
verify(selectedAdmin).createAndUpdateSubscriptionGroupConfig(
org.mockito.ArgumentMatchers.eq("10.0.0.1:10911"), any());
verify(adminExt,
never()).createAndUpdateSubscriptionGroupConfig(anyString(), any());
+ ArgumentCaptor<LambdaQueryWrapper<RmqGroup>> captor =
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+ verify(groupMapper).selectOne(captor.capture());
+ assertThat(captor.getValue().getSqlSegment()).contains("cluster_id",
"instance_id", "name");
}
@Test
@@ -448,7 +451,7 @@ class RocketMQAdminClientImplTest {
verify(adminExt, never()).deleteSubscriptionGroup(anyString(),
anyString(), org.mockito.ArgumentMatchers.anyBoolean());
ArgumentCaptor<LambdaQueryWrapper<RmqGroup>> captor =
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(groupMapper).delete(captor.capture());
- assertThat(captor.getValue().getSqlSegment()).contains("cluster_id",
"name");
+ assertThat(captor.getValue().getSqlSegment()).contains("cluster_id",
"instance_id", "name");
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index e6d5da09c..f74e81cc1 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -16,6 +16,10 @@
*/
package org.apache.rocketmq.studio.provider.apache;
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
@@ -38,6 +42,7 @@ import
org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
import org.apache.rocketmq.studio.instance.topic.TopicConsumerPageVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.persistence.entity.RmqGroup;
+import org.apache.rocketmq.studio.persistence.entity.RmqTopic;
import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
import org.apache.rocketmq.studio.persistence.mapper.RmqTopicMapper;
import org.junit.jupiter.api.Test;
@@ -59,6 +64,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.lenient;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.never;
@@ -163,6 +169,34 @@ class RocketMQMetadataProviderTest {
verifyNoInteractions(runtimeAdminClientResolver);
}
+ @Test
+ void listTopicsShouldScopeDatabaseQueryToSelectedInstance() {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+ when(topicMapper.selectList(any())).thenReturn(List.of());
+ RocketMQMetadataProvider provider = newProvider();
+
+ assertThat(provider.listTopics("instance-a", "cluster-1", "NORMAL",
"orders")).isEmpty();
+
+
org.mockito.ArgumentCaptor<LambdaQueryWrapper<org.apache.rocketmq.studio.persistence.entity.RmqTopic>>
captor =
+ org.mockito.ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+ verify(topicMapper).selectList(captor.capture());
+ assertThat(captor.getValue().getSqlSegment()).contains("instance_id",
"cluster_id");
+ }
+
+ @Test
+ void listConsumerGroupsShouldScopeDatabaseQueryToSelectedInstance() {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqGroup.class);
+ when(groupMapper.selectList(any())).thenReturn(List.of());
+ RocketMQMetadataProvider provider = newProvider();
+
+ assertThat(provider.listConsumerGroups("instance-a", "cluster-1",
"orders")).isEmpty();
+
+ org.mockito.ArgumentCaptor<LambdaQueryWrapper<RmqGroup>> captor =
+ org.mockito.ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+ verify(groupMapper, times(1)).selectList(captor.capture());
+ assertThat(captor.getValue().getSqlSegment()).contains("instance_id",
"cluster_id");
+ }
+
@Test
void getTopicRoutesShouldUseSelectedInstanceRuntimeClient() {
List<BrokerRouteVO> routes =
List.of(BrokerRouteVO.builder().brokerName("broker-a").build());
