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 ab778e7c1 feat(instance): scope topic and group uniqueness to the
nameserver domain (#7353)
ab778e7c1 is described below
commit ab778e7c12855c4d8a49186320d176fd3e8c1170
Author: lizhimins <[email protected]>
AuthorDate: Mon Oct 26 14:59:04 2026 +0800
feat(instance): scope topic and group uniqueness to the nameserver domain
(#7353)
Bind each APACHE instance to a nameserver registry entry at creation and
derive a
NOT NULL ownership domain from it (ns:<nameserver_id>,
cloud:<cloudInstanceId>),
so a topic or group name is unique per (domain, broker cluster) instead of
per
deployment. The ownership guard becomes slot based: same name on another
cluster
of the domain coexists, the owning instance re-creates as an update, and a
foreign
owner is reported with the create rejected.
Rewrite docs/studio-instance-model-spec.md to the domain wording: the spec
still
claimed deployment-wide global uniqueness and a name-only unique key, which
is what
this change replaces.
---
docs/studio-instance-model-spec.md | 98 ++++++---
.../nameserver/NameserverRegistryService.java | 20 ++
.../studio/instance/CreateInstanceDTO.java | 3 +
.../rocketmq/studio/instance/InstanceService.java | 24 +-
.../rocketmq/studio/instance/InstanceVO.java | 2 +
.../instance/MybatisPlusInstanceRepository.java | 5 +
.../studio/instance/ResourceOwnershipGuard.java | 150 ++++++++++---
.../instance/ResourceOwnershipSchemaMigration.java | 241 +++++++++++++++------
.../studio/instance/UpdateInstanceDTO.java | 3 +
.../studio/instance/message/MessageService.java | 2 +-
.../studio/instance/topic/LiteTopicService.java | 2 +-
.../studio/instance/topic/MetadataService.java | 6 +-
.../studio/persistence/entity/RmqGroup.java | 2 +
.../studio/persistence/entity/RmqInstance.java | 2 +
.../studio/persistence/entity/RmqTopic.java | 2 +
.../provider/apache/RocketMQAdminClientImpl.java | 128 +++++++----
.../provider/apache/RocketMQLiteTopicProvider.java | 4 +-
.../provider/apache/RocketMQMessageProvider.java | 8 +-
server/src/main/resources/db/schema.sql | 14 +-
.../studio/instance/InstanceServiceTest.java | 87 +++-----
.../MybatisPlusInstanceRepositoryTest.java | 3 +
.../instance/ResourceOwnershipGuardTest.java | 142 ++++++++----
.../instance/topic/LiteTopicServiceTest.java | 2 +-
.../studio/instance/topic/MetadataServiceTest.java | 32 +--
.../apache/RocketMQAdminClientImplTest.java | 41 +++-
.../apache/RocketMQLiteTopicProviderTest.java | 7 +-
.../apache/RocketMQMessageProviderTest.java | 6 +-
27 files changed, 716 insertions(+), 320 deletions(-)
diff --git a/docs/studio-instance-model-spec.md
b/docs/studio-instance-model-spec.md
index 6cbd50961..031a64dae 100644
--- a/docs/studio-instance-model-spec.md
+++ b/docs/studio-instance-model-spec.md
@@ -1,48 +1,94 @@
-# Studio 实例模型与 Topic/Group 全局唯一规范
+# Studio 实例模型与 Topic/Group 唯一性域规范
-> 状态：**规范已定**。当前代码与本规范不一致（DB 唯一键仍为 `(cluster_id, instance_id, name)`），落地项见
§5，未排期。
+> 状态：**已落地**。资源身份由「名字全局唯一」改为「按唯一性域 + broker 集群」，DB 唯一键为
+> `uk_topic_domain (domain, cluster_id, name)` / `uk_group_domain (domain,
cluster_id, name)`，
+> 存量库由 `ResourceOwnershipSchemaMigration` 在启动时回填并换键（见 §5）。本文取代此前的
+> 「Topic/Group 全局唯一」口径。
>
-> TL;DR (EN): One physical cluster may host multiple Studio instances;
`instanceId` is a Studio-side ownership concept only — open-source RocketMQ has
no namespace/instanceId. Topic and group names are **globally unique within one
Studio deployment**, even across multiple nameservers/clusters. Environments
that need identical resource names (dev/staging/prod) must run **separate
Studio deployments**.
+> TL;DR (EN): One physical cluster may host multiple Studio instances;
`instanceId` is a Studio-side
+> ownership concept only — open-source RocketMQ has no namespace/instanceId. A
topic or group name is
+> unique **per uniqueness domain and broker cluster**, where the domain is the
NameServer registry entry
+> an open-source instance is bound to (`ns:<nameserver_id>`) or the cloud
instance itself
+> (`cloud:<cloudInstanceId>`). Instances bound to different NameServer
clusters may therefore reuse the
+> same names within one Studio deployment, while two instances sharing one
NameServer and one broker
+> cluster still cannot claim the same slot.
## 1. 层级模型
```
-nameserver（接入点，物理） ← 多个 cluster 可共享
+nameserver（接入点，物理；rmq_nameserver 一行） ← 多个 cluster 可共享
└── cluster（物理集群） ← 1 个 cluster 可挂多个 instance
└── instance（Studio 逻辑管理单元，instanceId = rmq_instance.name 全局唯一）
+ └── topic / group（资源身份 = domain + cluster_id + name）
```
- 1 个 cluster 上可以有**多个 instance**（1:N）；每个 instance 由唯一 `instanceId` 标识。
- 多个 cluster 可能**共享同一 nameserver**。
-- **instanceId 只是 Studio 侧概念**：开源 RocketMQ 运行时没有 namespace / instanceId——一个
nameserver 之下，topic/group 集合是确定性的、扁平的。
+- **开源（APACHE）实例在创建时绑定一个 nameserver 注册项**（`rmq_instance.nameserver_id`，创建后
+ 不可更改）；直连模式的接入地址必须始终指向该集群。云厂商实例的域由 `cloud_instance_id` 给出，
+ 不需要绑定 nameserver。
+- **instanceId 只是 Studio 侧概念**：开源 RocketMQ 运行时没有 namespace / instanceId——一个
+ nameserver 之下，topic/group 集合是确定性的、扁平的。
-## 2. Topic/Group 全局唯一（核心约定）
+## 2. 唯一性域与 slot（核心约定）
-- 一个 Studio 部署内，topic 名与 group 名各自**全局唯一**：数据库唯一键直接加在 `name` 上（替代现行
`(cluster_id, instance_id, name)` 三列 UK）。
-- 一个名字只有一条管理记录：`instance_id` 仅表达**归属实例**（管理视图），`cluster_id`
仅表达物理集群引用——两者都不是资源身份的组成部分。
-- **即使用户管理多套 nameserver / 多个物理集群，名字同样要求全局唯一**（不按 cluster 隔离）：
- - 管理考量：检索、审计、告警与 MCP/CLI 工具中资源语义单一，杜绝「同名不同集群」的歧义。
- - 技术考量：共享 nameserver 时，路由按 topic 名注册与查询（`TopicRouteData` 按名聚合、不按 cluster
隔离），跨集群同名 topic 本身就会造成路由混杂——全局唯一从源头消除该隐患。
+- **域（domain）**：`ns:<nameserver_id>`（开源实例）或 `cloud:<cloudInstanceId>`（云实例），落库为
+ `rmq_instance_topic.domain` / `rmq_instance_group.domain`（NOT
NULL）。域表达「这个名字在哪一套
+ 接入点下才可能被路由到」。
+- **slot = (domain, cluster_id, name)**：唯一键直接建在这三列上，**一个 slot 只归属一个实例**。
+ 一个名字在一个域内可以有多个 slot（同域不同 broker cluster 各一条管理记录）。
+- 由此得到三条边界：
+ - **跨域同名互不影响**——日常 / 预发 / 线上各自绑定自己的 nameserver，就能在一个 Studio 内共存。
+ - **同域跨 cluster 同名可共存**——路由本身按 cluster 隔离，各 cluster 上是各自独立的物理资源。
+ - **同域同 cluster 同名 = 同一个物理资源**，归属唯一；他人已占有则拒绝（见 §4）。
+- `instance_id` 仅表达**归属实例**（管理视图），不是资源身份的组成部分。
+- 未绑定 nameserver 的开源实例无法确定域：`ResourceOwnershipGuard.domainOf` 抛 409，提示带
+ nameserver 重新创建该实例（老实例由启动迁移自动补绑，见 §5）。
- **对外参数命名**（MCP tools / REST / CLI 一体适用）：实例维度入参一律 **`instanceId`**，禁止以
`cluster` 作为实例参数名——`cluster`/`clusterId` 仅保留物理集群语义（如
`cluster.list`、`TopicVO.clusterId`）；`instanceId`
为**正式必填参数、无默认注入设计**，由调用方显式传入（工具 schema 与参数表必须显式列出，不得以「由 context
提供」等理由省略）；资源名入参一律 **`topicName` / `groupName`**，单资源操作为精确匹配，list
类工具以其为模糊查询参数（替代泛化的 `search`）；CLI/MCP **无 namespace 概念**——工具入/出参一律不暴露
`namespace`（开源链路无 `ns%topic` 物理前缀，namespace 仅为展示字段，从工具中删除；REST/VO 与 CSV 导出不受影响）。
+- `clusterId` 只在「域内多个 broker cluster 都有该名字」时才是必填的消歧参数，其余情况由后端解析
+ （见 §4 第 2 条），不把它变成所有接口的常规必填项。
-## 3. 同名需求 = 部署多套 Studio
+## 3. 同名需求 = 绑不同的 nameserver
-- 用户确需同名 topic/group（典型：**日常 / 预发 / 线上**三套环境复用同一批资源名）时，正确做法是**部署多套独立
Studio**（每环境一套），而不是在一个 Studio 内容忍重名。
-- 一个 Studio = 一个环境的治理域，域内全局唯一天然成立。
+- 用户确需同名 topic/group（典型：**日常 / 预发 / 线上**三套环境复用同一批资源名）时，正确做法是
+ **给每套环境的实例绑定各自的 nameserver 集群**——域不同，名字自然互不冲突。
+- 部署多套 Studio 仍然可行，但不再是同名需求的唯一手段。
+- 一个 Studio = 一个治理面板，面板内可以并列多套环境；真正的冲突只发生在「同域 + 同 broker
+ cluster + 同名」，而那本来就是同一个物理资源。
## 4. 行为约束（由规范导出，REST / MCP / CLI 一体适用）
-1. **create / import**：先按 `name` 查重——名字已存在且归属其他实例 → 拒绝（409，提示归属实例）；归属本实例 → 幂等
upsert（create==update 语义不变）。import **不得静默覆盖**物理配置，同名冲突逐条记入 failures。
-2. **update / delete /
消息与位点操作**：仅归属实例可执行；跨实例请求拒绝（409，提示归属实例）。归属转移暂不提供（如确有需求另行设计「转移归属」功能）。
-3. **list / describe**：管理视图——仅展示归属该实例的资源；**物理可见 ≠ 归属**（同一 cluster
新建的实例不会自动看到既有 topic）。既有集群资源经 import 按名进入某实例视图（受第 1 条查重约束）。
-4. **delete**：删除物理资源 + 唯一管理记录；不存在多实例「僵尸记录」问题（一实例删除、另一实例仍显示）——该问题随三列 UK 废除而根除。
+1. **create / import**：目标 cluster 取「本实例在该域内已有的同名注册」，没有则用实例默认 cluster
+ （`createTarget`）；**其他 cluster 上他人的同名注册不阻塞创建**。解析出的 slot 归属他实例 →
+ 拒绝（409，提示归属实例）；归属本实例 → 幂等 upsert（create==update 语义不变）。import
+ **不得静默覆盖**物理配置，同名冲突逐条记入 failures。
+2. **配置类变更（update / delete / topic 与 group 配置）**：必须有唯一物理目标（`checkSlot`）。
+ `clusterId` 为空时，该名字在域内只有一条注册就用它，有多条则拒绝（409，提示需要指定物理目标）；
+ 仅归属实例可执行，跨实例请求拒绝（409，提示归属实例）。归属转移暂不提供（如确有需求另行设计
+ 「转移归属」功能）。
+3. **按名路由的操作（发消息、查消息、重置位点等）**：走域级归属（`checkOwnerInDomain`）——本实例
+ 自己的注册优先命中；域内只有他人注册时拒绝（409）。
+4. **list / describe**：管理视图——仅展示归属该实例的资源；**物理可见 ≠ 归属**（同一 cluster 新建的
+ 实例不会自动看到既有 topic）。既有集群资源经 import 按 slot 进入某实例视图（受第 1 条约束）。
+5. **delete**：删除物理资源 + 该 slot 的管理记录；同域其他 cluster、其他域的同名注册不受影响。
+ 归属一旦提交，不因后续远端操作失败而释放（避免半删状态被他人抢占）。
-## 5. 落地项（待办，未排期）
+## 5. 实现位置
-- `server/src/main/resources/db/schema.sql`：
- - `uk_cluster_instance_topic (cluster_id, instance_id, name)` →
`uk_topic_name (name)`；
- - `uk_cluster_instance_group (cluster_id, instance_id, name)` →
`uk_group_name (name)`；
- - `cluster_id` / `instance_id` 列保留（归属与物理集群引用），维持普通索引（`idx_topic_instance` 等）。
-- `RocketMQAdminClientImpl.createTopic / deleteTopic`（及 group 对应方法）：DB
查重与删除条件由 `(cluster_id, instance_id, name)` 改为按 `name`，加归属校验与 409
冲突错误；`MetadataService.importTopics` 随 `createTopic` 生效。
-- **存量数据迁移**：迁移脚本先检测跨实例同名行；存在即停下报告、由人工合并或裁决归属，禁止自动丢弃。
-- 测试与文档：补充归属冲突 / 跨实例拒绝用例。
+-
`server/src/main/resources/db/schema.sql`：`rmq_instance.nameserver_id`（应用层校验存在性，不加外键）、
+ 两张资源表的 `domain` 列、`uk_topic_domain` / `uk_group_domain` 与 `idx_*_domain`。
+- `ResourceOwnershipGuard`：`domainOf`（算域）、`createTarget` / `checkSlot` /
`checkOwnerInDomain` /
+ `slots`（解析 slot 与归属）、`write`（带行锁落归属）；云资源尚未全局登记，写操作直接拒绝而不把
+ OpenAPI 可见性当归属。
+- `ResourceOwnershipSchemaMigration`：启动时补列 → 逐实例解析域（nameserver 注册表里没有的接入地址
+ 自动注册一行，云实例缺 `cloud_instance_id` 则拒绝启动）→ 回填资源行 `domain`（归属为空的历史行按
+ `cluster_id` 兜底，仍无法判定则拒绝启动）→ 校验同 slot 无重名（有则拒绝启动、要求人工裁决）→
+ **先建 `uk_*_domain` 再删旧唯一键**，中途失败不丢失既有约束。
+- `InstanceService` / `CreateInstanceDTO` / `UpdateInstanceDTO` / `InstanceVO`
/
+ `MybatisPlusInstanceRepository`：`nameserverId`
创建必填、更新不可改；`NameserverRegistryService`
+ 提供选择器所需的注册项与存在性校验。
+- 调用点：`RocketMQAdminClientImpl`（topic / group
的创建、删除、配置）、`MetadataService`（import）、
+ `LiteTopicService`、`MessageService`、`RocketMQMessageProvider` /
`RocketMQLiteTopicProvider`。
+- 前端：实例页的 nameserver 选择器与列；创建实例 / Topic / Group 三处表单的问号弹窗
+ （`ResourceRuleHelp` + `ResourceOwnershipTopology`）——把域边界、归属拓扑与四条归属规则讲在
+ 「创建动作发生的地方」，不设独立说明页。
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryService.java
index 516f6bac0..c964291ae 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryService.java
@@ -126,6 +126,26 @@ public class NameserverRegistryService {
return new BusinessException(409, "NameServer registry name already
exists: " + name);
}
+ public NameserverRegistryVO requireExisting(long id) {
+ RmqNameserver entity = nameserverMapper.selectById(id);
+ if (entity == null) {
+ throw new BusinessException(404, "NameServer registry entry not
found: " + id);
+ }
+ return toVO(entity);
+ }
+
+ public String addressOf(long id) {
+ return requireExisting(id).getNamesrvAddr();
+ }
+
+ public String nameOf(Long id) {
+ if (id == null) {
+ return null;
+ }
+ RmqNameserver entity = nameserverMapper.selectById(id);
+ return entity == null ? null : entity.getName();
+ }
+
public void delete(Long id) {
if (nameserverMapper.selectById(id) == null) {
throw new BusinessException(404, "NameServer registry entry not
found: " + id);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/CreateInstanceDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/CreateInstanceDTO.java
index 8b3497fe9..cdb8d33ed 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/CreateInstanceDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/CreateInstanceDTO.java
@@ -41,6 +41,8 @@ public class CreateInstanceDTO {
private String regionId;
+ private Long nameserverId;
+
public InstanceVO toInstanceVO() {
InstanceVO vo = InstanceVO.builder()
.name(name)
@@ -52,6 +54,7 @@ public class CreateInstanceDTO {
.credentialId(credentialId)
.adminCredentialRef(adminCredentialRef)
.regionId(regionId)
+ .nameserverId(nameserverId)
.build();
return vo;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
index 462e188d6..6fce8cae2 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
@@ -22,6 +22,8 @@ import org.springframework.util.StringUtils;
import
org.apache.rocketmq.studio.provider.credential.CloudCredentialRepository;
import org.apache.rocketmq.studio.provider.credential.CloudCredentialVO;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryService;
+import org.apache.rocketmq.studio.cluster.nameserver.NamesrvAddrParser;
import org.apache.rocketmq.studio.cluster.broker.MqClientPool;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
@@ -77,6 +79,7 @@ public class InstanceService {
private final CacheManager cacheManager;
private final RegionNames regionNames;
private final ResourceOwnershipGuard ownershipGuard;
+ private final NameserverRegistryService nameserverRegistry;
// @Lazy self-injection: Spring AOP proxies intercept @Transactional calls
only when they
// originate from outside the bean. Calling deleteInstance() directly from
within this class
@@ -510,6 +513,10 @@ public class InstanceService {
if (instance.getType() == InstanceType.CLOUD) {
throw new BusinessException(400, "CLOUD type is reserved for
vendor-managed instances");
}
+ if (instance.getNameserverId() == null || instance.getNameserverId()
<= 0) {
+ throw new BusinessException(400, "nameserverId is required");
+ }
+ nameserverRegistry.requireExisting(instance.getNameserverId());
}
/**
@@ -654,8 +661,23 @@ public class InstanceService {
}
updated.setType(instance.getType());
}
+ if (instance.getNameserverId() != null
+ &&
!instance.getNameserverId().equals(existing.getNameserverId())) {
+ throw new BusinessException(400, "The bound NameServer cluster
cannot be changed after creation");
+ }
if (instance.getEndpoint() != null) {
-
updated.setEndpoint(requireValidEndpoint(instance.getEndpoint()));
+ String normalized =
requireValidEndpoint(instance.getEndpoint());
+ // A DIRECT endpoint is the NameServer address itself, so it
must keep matching the binding;
+ // proxy endpoints address the proxy tier and stay free to
move inside the same domain.
+ if (existing.getType() == InstanceType.DIRECT &&
existing.getNameserverId() != null
+ &&
!NamesrvAddrParser.normalize(nameserverRegistry.addressOf(existing.getNameserverId()))
+
.equals(NamesrvAddrParser.normalize(normalized))) {
+ throw new BusinessException(400,
+ "The endpoint must stay inside the bound
NameServer cluster: " + existing.getName());
+ }
+ // A DIRECT endpoint is stored in the same normalized form as
the registry address it binds to.
+ updated.setEndpoint(existing.getType() == InstanceType.DIRECT
&& existing.getNameserverId() != null
+ ? NamesrvAddrParser.normalize(normalized) :
normalized);
}
}
if (instance.getRemark() != null) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceVO.java
index 60a14d155..f38f83664 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceVO.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceVO.java
@@ -42,6 +42,8 @@ public class InstanceVO extends BaseEntity {
private String adminCredentialRef;
private String regionId;
private String regionName;
+ private Long nameserverId;
+ private String nameserverName;
private int topicCount;
private int consumerGroupCount;
@Builder.Default
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 7fcd1a33f..92ad16c60 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
@@ -27,6 +27,7 @@ import
org.apache.rocketmq.studio.persistence.entity.RmqInstance;
import org.apache.rocketmq.studio.persistence.entity.RmqTopic;
import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
import org.apache.rocketmq.studio.persistence.mapper.RmqInstanceMapper;
+import org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryService;
import org.apache.rocketmq.studio.persistence.mapper.RmqTopicMapper;
import org.springframework.stereotype.Repository;
import lombok.RequiredArgsConstructor;
@@ -42,6 +43,7 @@ public class MybatisPlusInstanceRepository implements
InstanceRepository {
private final RmqInstanceMapper instanceMapper;
private final RmqTopicMapper topicMapper;
private final RmqGroupMapper groupMapper;
+ private final NameserverRegistryService nameserverRegistry;
@Override
public List<InstanceVO> findAll() {
@@ -176,7 +178,9 @@ public class MybatisPlusInstanceRepository implements
InstanceRepository {
.credentialId(entity.getCredentialId())
.adminCredentialRef(entity.getAdminCredentialRef())
.regionId(entity.getRegionId())
+ .nameserverId(entity.getNameserverId())
.build();
+
vo.setNameserverName(nameserverRegistry.nameOf(entity.getNameserverId()));
vo.setId(entity.getId());
vo.setGmtCreate(entity.getGmtCreate());
vo.setGmtModified(entity.getGmtModified());
@@ -223,6 +227,7 @@ public class MybatisPlusInstanceRepository implements
InstanceRepository {
entity.setCredentialId(vo.getCredentialId());
entity.setAdminCredentialRef(vo.getAdminCredentialRef());
entity.setRegionId(vo.getRegionId());
+ entity.setNameserverId(vo.getNameserverId());
entity.setGmtCreate(vo.getGmtCreate());
entity.setGmtModified(vo.getGmtModified());
return entity;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
index 651d8ab2f..b311253d7 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
@@ -36,11 +36,19 @@ import java.util.List;
import java.util.Objects;
import java.util.function.Supplier;
-/** Global ownership with database row locks; a committed claim is never
released by remote operation failures. */
+/**
+ * Slot ownership with database row locks. A slot is (domain, cluster, name):
the domain is the
+ * NameServer cluster an instance is bound to (or the cloud instance itself),
so the same name may
+ * live in different domains and on different broker clusters of one domain,
while one slot belongs
+ * to exactly one instance. A committed claim is never released by remote
operation failures.
+ */
@Component
@DependsOn("resourceOwnershipSchemaMigration")
@RequiredArgsConstructor
public class ResourceOwnershipGuard {
+ private static final String NAMESERVER_DOMAIN_PREFIX = "ns:";
+ private static final String CLOUD_DOMAIN_PREFIX = "cloud:";
+
private final JdbcTemplate jdbc;
private final PlatformTransactionManager transactionManager;
private final InstanceResolver instanceResolver;
@@ -54,7 +62,8 @@ public class ResourceOwnershipGuard {
public record Resource(Kind kind, String name) { }
- public record Ownership(long id, String name, String instanceId, String
clusterId, String topicType) { }
+ public record Ownership(long id, String name, String instanceId, String
clusterId, String topicType,
+ String domain) { }
public InstanceVO requireInstance(String identifier) {
String id = requireText(identifier, "instanceId");
@@ -69,6 +78,27 @@ public class ResourceOwnershipGuard {
return value.trim();
}
+ public static String nameserverDomain(long nameserverId) {
+ return NAMESERVER_DOMAIN_PREFIX + nameserverId;
+ }
+
+ public static String cloudDomain(String cloudInstanceId) {
+ return CLOUD_DOMAIN_PREFIX + requireText(cloudInstanceId,
"cloudInstanceId");
+ }
+
+ /** The uniqueness domain an instance lives in; open-source instances must
be bound to a NameServer. */
+ public String domainOf(InstanceVO instance) {
+ if (instance.getVendor() != null && instance.getVendor() !=
InstanceVendor.APACHE) {
+ return cloudDomain(instance.getCloudInstanceId());
+ }
+ if (instance.getNameserverId() == null || instance.getNameserverId()
<= 0) {
+ throw new BusinessException(409,
+ "Instance is not bound to a NameServer cluster; re-create
it with a NameServer selected: "
+ + instance.getName());
+ }
+ return nameserverDomain(instance.getNameserverId());
+ }
+
/** Derived topics borrow the group ownership only; no regular topic
registration is created. */
public Resource topicResource(String name) {
String topic = requireText(name, "topicName");
@@ -85,8 +115,50 @@ public class ResourceOwnershipGuard {
return new Resource(Kind.TOPIC, topic);
}
- public Ownership check(InstanceVO instance, Resource resource, boolean
required) {
- return checkOwner(instance, resource, read(resource, false), required);
+ /**
+ * Resolves the slot of a configuration mutation. A null cluster accepts
the single registration of
+ * the name inside the domain and rejects an ambiguous one, so the caller
learns the physical target.
+ */
+ public Ownership checkSlot(InstanceVO instance, Resource resource, String
cluster, boolean required) {
+ List<Ownership> rows = read(domainOf(instance), resource, cluster,
false);
+ if (rows.size() > 1) {
+ throw new BusinessException(409, resource.name()
+ + " exists in multiple Broker clusters; a physical target
is required for configuration changes");
+ }
+ return checkOwner(instance, resource, rows.isEmpty() ? null :
rows.getFirst(), required);
+ }
+
+ /**
+ * Domain-level ownership for operations that route by name instead of
mutating a registration
+ * (sending messages, resetting offsets): the caller's own registration
wins, a foreign one rejects.
+ */
+ public Ownership checkOwnerInDomain(InstanceVO instance, Resource
resource, boolean required) {
+ List<Ownership> rows = read(domainOf(instance), resource, null, false);
+ Ownership own = rows.stream()
+ .filter(row -> instance.getName().equals(row.instanceId()))
+ .findFirst()
+ .orElse(null);
+ if (own != null) {
+ return own;
+ }
+ return checkOwner(instance, resource, rows.isEmpty() ? null :
rows.getFirst(), required);
+ }
+
+ /** Every registration of the name inside the instance domain, one per
broker cluster. */
+ public List<Ownership> slots(InstanceVO instance, Resource resource) {
+ return read(domainOf(instance), resource, null, false);
+ }
+
+ /**
+ * Picks the cluster a creation should target: the caller's existing
registration when there is one,
+ * otherwise the instance default. Foreign registrations on other clusters
do not block a creation;
+ * {@link #write} rejects the request only when the resolved slot itself
is owned by someone else.
+ */
+ public Ownership createTarget(InstanceVO instance, Resource resource) {
+ return read(domainOf(instance), resource, null, false).stream()
+ .filter(row -> instance.getName().equals(row.instanceId()))
+ .findFirst()
+ .orElse(null);
}
/** Cloud resources are not registered globally yet; reject writes instead
of treating OpenAPI visibility as ownership. */
@@ -103,10 +175,10 @@ public class ResourceOwnershipGuard {
throw new BusinessException(400, "resourceName is required");
}
if (topic != null) {
- check(instance, topicResource(topic), false);
+ checkSlot(instance, topicResource(topic), null, false);
}
if (group != null) {
- check(instance, new Resource(Kind.GROUP, requireText(group,
"groupName")), false);
+ checkSlot(instance, new Resource(Kind.GROUP, requireText(group,
"groupName")), null, false);
}
throw new BusinessException(501, "Cloud resources are not registered
for global ownership yet; resource writes are not allowed");
}
@@ -114,36 +186,39 @@ public class ResourceOwnershipGuard {
public <T> T write(InstanceVO instance, Resource resource, String cluster,
boolean create,
String topicType, Supplier<T> action) {
requireSupportedProvider(instance);
+ String domain = domainOf(instance);
+ String target = requireText(cluster, "clusterId");
// Commit the claim first; later RPC, DB updates or process crashes
must not let other instances take over.
try {
newTransaction().executeWithoutResult(status -> {
lockInstance(instance);
- Ownership existing = checkOwner(instance, resource,
read(resource, true), !create);
+ Ownership existing = checkOwner(instance, resource,
+ read(domain, resource, target,
true).stream().findFirst().orElse(null), !create);
if (existing == null) {
if (resource.kind() == Kind.TOPIC) {
- jdbc.update("INSERT INTO rmq_instance_topic (name,
instance_id, cluster_id, status, topic_type)"
- + " VALUES (?, ?, ?, 'PENDING', ?)",
resource.name(), instance.getName(), cluster,
- topicType == null ? "NORMAL" : topicType);
+ jdbc.update("INSERT INTO rmq_instance_topic (name,
instance_id, cluster_id, domain, status, topic_type)"
+ + " VALUES (?, ?, ?, ?, 'PENDING',
?)", resource.name(), instance.getName(), target,
+ domain, topicType == null ? "NORMAL" :
topicType);
} else {
- jdbc.update("INSERT INTO rmq_instance_group (name,
instance_id, cluster_id, status)"
- + " VALUES (?, ?, ?, 'PENDING')",
resource.name(), instance.getName(), cluster);
+ jdbc.update("INSERT INTO rmq_instance_group (name,
instance_id, cluster_id, domain, status)"
+ + " VALUES (?, ?, ?, ?, 'PENDING')",
resource.name(), instance.getName(), target,
+ domain);
}
} else {
- requireCluster(existing, cluster);
requireTopicType(existing, topicType);
}
});
} catch (DuplicateKeyException conflict) {
// The unique-key race happens before any RPC; never continue
remote writes after a unique-key failure.
- check(instance, resource, true);
+ checkSlot(instance, resource, target, true);
throw new BusinessException(409, "Resource is being registered
concurrently, please retry: " + resource.name());
} catch (PessimisticLockingFailureException conflict) {
throw new BusinessException(409, "Resource registration lock
conflict, please retry: " + resource.name());
}
return newTransaction().execute(status -> {
lockInstance(instance);
- Ownership existing = checkOwner(instance, resource, read(resource,
true), true);
- requireCluster(existing, cluster);
+ Ownership existing = checkOwner(instance, resource,
+ read(domain, resource, target,
true).stream().findFirst().orElse(null), true);
requireTopicType(existing, topicType);
return action.get();
});
@@ -156,7 +231,7 @@ public class ResourceOwnershipGuard {
lockInstance(instance);
resources.stream().distinct().sorted(Comparator.comparing((Resource r) ->
r.kind().name())
.thenComparing(Resource::name)).forEach(resource ->
- checkOwner(instance, resource, read(resource, true),
true));
+ checkOwnerInDomain(instance, resource, true));
return action.get();
});
}
@@ -220,8 +295,8 @@ public class ResourceOwnershipGuard {
lockIdentity(instance.getName());
if (instance.getId() == null) {
// A virtual instance must not be silently replaced by a
registered instance with the same name.
- if (!jdbc.query("SELECT id FROM rmq_instance WHERE name = ? FOR
UPDATE",
- (rs, row) -> rs.getLong(1), instance.getName()).isEmpty())
{
+ if (!jdbc.query("SELECT name FROM rmq_instance WHERE name = ? FOR
UPDATE",
+ (rs, row) -> rs.getString(1),
instance.getName()).isEmpty()) {
throw new BusinessException(409, "Virtual instance is shadowed
by a registered instance; please re-confirm ownership");
}
return;
@@ -239,16 +314,27 @@ public class ResourceOwnershipGuard {
}
}
- private Ownership read(Resource resource, boolean lock) {
+ private List<Ownership> read(String domain, Resource resource, String
cluster, boolean lock) {
String type = resource.kind() == Kind.TOPIC ? "topic_type" : "NULL";
- List<Ownership> rows = jdbc.query("SELECT id, name, instance_id,
cluster_id, " + type
- + " FROM " + resource.kind().table + " WHERE name = ?"
+ (lock ? " FOR UPDATE" : ""),
- (rs, row) -> new Ownership(rs.getLong(1), rs.getString(2),
rs.getString(3),
- rs.getString(4), rs.getString(5)), resource.name());
- if (rows.size() > 1) {
- throw new BusinessException(409, "Resource has duplicate ownership
records and needs manual resolution: " + resource.name());
+ StringBuilder sql = new StringBuilder("SELECT id, name, instance_id,
cluster_id, " + type + ", domain FROM "
+ + resource.kind().table + " WHERE domain = ? AND name = ?");
+ if (cluster != null) {
+ sql.append(" AND cluster_id = ?");
+ }
+ if (lock) {
+ sql.append(" FOR UPDATE");
}
- return rows.isEmpty() ? null : rows.getFirst();
+ String query = sql.toString();
+ return jdbc.query(connection -> {
+ var prepared = connection.prepareStatement(query);
+ prepared.setString(1, domain);
+ prepared.setString(2, resource.name());
+ if (cluster != null) {
+ prepared.setString(3, cluster);
+ }
+ return prepared;
+ }, (rs, row) -> new Ownership(rs.getLong(1), rs.getString(2),
rs.getString(3),
+ rs.getString(4), rs.getString(5), rs.getString(6)));
}
private Ownership checkOwner(InstanceVO instance, Resource resource,
Ownership owner, boolean required) {
@@ -258,21 +344,13 @@ public class ResourceOwnershipGuard {
}
return null;
}
- boolean legacyVirtual = !StringUtils.hasText(owner.instanceId()) &&
instance.getId() == null
- && instance.getName().equals(owner.clusterId());
- if (!instance.getName().equals(owner.instanceId()) && !legacyVirtual) {
+ if (!instance.getName().equals(owner.instanceId())) {
throw new BusinessException(409, "Resource " + resource.name() + "
belongs to instance: "
+ (StringUtils.hasText(owner.instanceId()) ?
owner.instanceId() : "unknown (legacy record, manual confirmation required)"));
}
return owner;
}
- private void requireCluster(Ownership owner, String cluster) {
- if (!Objects.equals(owner.clusterId(), cluster)) {
- throw new BusinessException(409, "Resource physical cluster does
not match the write target: " + owner.name());
- }
- }
-
private void requireTopicType(Ownership owner, String requested) {
if (requested != null && owner.topicType() != null &&
!requested.equals(owner.topicType())) {
throw new BusinessException(400, "topic message type is
immutable");
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
index 9efdff712..c09e65293 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
@@ -17,7 +17,7 @@
package org.apache.rocketmq.studio.instance;
import lombok.RequiredArgsConstructor;
-import
org.apache.rocketmq.studio.provider.apache.RocketMQDefaultClusterResolver;
+import org.apache.rocketmq.studio.cluster.nameserver.NamesrvAddrParser;
import org.springframework.beans.factory.InitializingBean;
import
org.springframework.boot.sql.init.dependency.DependsOnDatabaseInitialization;
import org.springframework.stereotype.Component;
@@ -25,16 +25,24 @@ import org.springframework.util.StringUtils;
import javax.sql.DataSource;
import java.sql.Connection;
+import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
-/** Validates the migration after SQL initialization and before write services
start; never deduplicates or rewrites ownership. */
+/**
+ * Moves resource uniqueness from the bare name to the (domain, cluster, name)
slot: adds the missing
+ * columns on already initialized databases, resolves every instance to its
domain (registering a
+ * NameServer row for endpoints the registry does not know yet), backfills the
domain of every resource
+ * row, refuses to start on an unresolvable row or a duplicated slot, and only
then swaps the unique keys.
+ */
@Component
@DependsOnDatabaseInitialization
@RequiredArgsConstructor
@@ -43,52 +51,22 @@ public class ResourceOwnershipSchemaMigration implements
InitializingBean {
private static final int INDEX_METADATA_ATTEMPTS = 5;
private final DataSource dataSource;
- private final RocketMQDefaultClusterResolver defaultClusters;
@Override
public void afterPropertiesSet() throws Exception {
try (Connection connection = dataSource.getConnection(); Statement
statement = connection.createStatement()) {
- // Both tables are pre-checked; no DDL runs on any conflict and no
record is discarded.
+ ensureColumns(connection, statement);
+ Map<String, String> domains = backfillInstances(connection);
for (String kind : List.of("topic", "group")) {
- String table = "rmq_instance_" + kind;
- try (ResultSet rows = statement.executeQuery("SELECT name,
COUNT(*) FROM " + table
- + " GROUP BY name HAVING COUNT(*) > 1")) {
- if (rows.next()) {
- throw new IllegalStateException("Global resource
uniqueness migration failed: duplicate name requires manual resolution: "
- + table + "/" + rows.getString(1));
- }
- }
+ backfillResources(connection, "rmq_instance_" + kind, domains);
}
- List<String> virtualNames = null;
for (String kind : List.of("topic", "group")) {
- try (ResultSet rows = statement.executeQuery("SELECT r.name,
r.instance_id, r.cluster_id, i.name, owner.name"
- + " FROM rmq_instance_" + kind + " r LEFT JOIN
rmq_instance i ON i.name = r.cluster_id"
- + " LEFT JOIN rmq_instance owner ON owner.name =
r.instance_id")) {
- while (rows.next()) {
- if (!StringUtils.hasText(rows.getString(1)) ||
!StringUtils.hasText(rows.getString(3))
- ||
!rows.getString(1).equals(rows.getString(1).trim())
- ||
!rows.getString(3).equals(rows.getString(3).trim())) {
- throw new IllegalStateException("Resource
ownership cannot be determined and requires manual confirmation: " +
rows.getString(1));
- }
- if (!StringUtils.hasText(rows.getString(2))) {
- if (virtualNames == null) {
- virtualNames = defaultClusters.names();
- }
- if (rows.getString(4) != null ||
!virtualNames.contains(rows.getString(3))) {
- throw new IllegalStateException("Legacy empty
instance_id has no trusted virtual cluster mapping and requires manual
confirmation: "
- + rows.getString(1));
- }
- } else if
(!rows.getString(2).equals(rows.getString(2).trim())) {
- throw new IllegalStateException("Instance
ownership identifier is not normalized and requires manual confirmation: " +
rows.getString(1));
- } else if (rows.getString(5) == null) {
- if (virtualNames == null) {
- virtualNames = defaultClusters.names();
- }
- if (!rows.getString(2).equals(rows.getString(3))
- ||
!virtualNames.contains(rows.getString(3))) {
- throw new IllegalStateException("Owning
instance of the resource does not exist and has no trusted virtual mapping: " +
rows.getString(1));
- }
- }
+ String table = "rmq_instance_" + kind;
+ try (ResultSet rows = statement.executeQuery("SELECT name FROM
" + table
+ + " GROUP BY domain, cluster_id, name HAVING COUNT(*)
> 1")) {
+ if (rows.next()) {
+ throw new IllegalStateException("Resource uniqueness
migration failed: the same name is registered"
+ + " twice in one slot and needs manual
resolution: " + table + "/" + rows.getString(1));
}
}
}
@@ -100,40 +78,181 @@ public class ResourceOwnershipSchemaMigration implements
InitializingBean {
+ " ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
for (String kind : List.of("topic", "group")) {
String table = "rmq_instance_" + kind;
- ensureIndex(connection, table, "uk_" + kind + "_name", "name",
true);
- ensureIndex(connection, table, "idx_" + kind + "_instance",
"instance_id", false);
- ensureIndex(connection, table, "idx_" + kind + "_cluster",
"cluster_id", false);
- // Create the global unique key before dropping the legacy
composite index; a mid-way failure keeps existing constraints.
- String legacy = "uk_cluster_instance_" + kind;
- if (indexes(connection, table).containsKey(legacy)) {
- try {
- statement.executeUpdate("ALTER TABLE " + table + "
DROP INDEX " + legacy);
- } catch (SQLException failure) {
- if (indexes(connection, table).containsKey(legacy)) {
- throw failure;
- }
- }
+ // Create the slot unique key before dropping the name-only
one; a mid-way failure keeps existing constraints.
+ ensureIndex(connection, table, "uk_" + kind + "_domain",
List.of("domain", "cluster_id", "name"), true);
+ ensureIndex(connection, table, "idx_" + kind + "_instance",
List.of("instance_id"), false);
+ ensureIndex(connection, table, "idx_" + kind + "_cluster",
List.of("cluster_id"), false);
+ ensureIndex(connection, table, "idx_" + kind + "_domain",
List.of("domain"), false);
+ dropIndex(connection, table, "uk_" + kind + "_name");
+ dropIndex(connection, table, "uk_cluster_instance_" + kind);
+ }
+ }
+ }
+
+ private void ensureColumns(Connection connection, Statement statement)
throws SQLException {
+ addColumnIfMissing(connection, statement, "rmq_instance",
"nameserver_id",
+ "bigint(20) unsigned COMMENT 'bound rmq_nameserver.id;
validated in the application layer, no foreign key'");
+ for (String kind : List.of("topic", "group")) {
+ addColumnIfMissing(connection, statement, "rmq_instance_" + kind,
"domain",
+ "VARCHAR(128) NOT NULL DEFAULT '' COMMENT 'uniqueness
domain: ns:<nameserver_id> or cloud:<cloud_instance_id>'");
+ }
+ }
+
+ private void addColumnIfMissing(Connection connection, Statement
statement, String table, String column,
+ String definition) throws SQLException {
+ if (columns(connection, table).contains(column)) {
+ return;
+ }
+ statement.executeUpdate("ALTER TABLE " + table + " ADD COLUMN " +
column + " " + definition);
+ }
+
+ private List<String> columns(Connection connection, String table) throws
SQLException {
+ List<String> columns = new ArrayList<>();
+ try (ResultSet rows =
connection.getMetaData().getColumns(connection.getCatalog(), null, table,
null)) {
+ while (rows.next()) {
+
columns.add(rows.getString("COLUMN_NAME").toLowerCase(Locale.ROOT));
+ }
+ }
+ return columns;
+ }
+
+ /** Resolves every instance to its domain, registering a NameServer row
for endpoints the registry does not know yet. */
+ private Map<String, String> backfillInstances(Connection connection)
throws SQLException {
+ Map<String, String> domains = new LinkedHashMap<>();
+ List<Long> ids = new ArrayList<>();
+ List<String> names = new ArrayList<>();
+ List<String> endpoints = new ArrayList<>();
+ List<String> vendors = new ArrayList<>();
+ List<String> cloudIds = new ArrayList<>();
+ List<Long> nameserverIds = new ArrayList<>();
+ try (Statement statement = connection.createStatement();
+ ResultSet rows = statement.executeQuery("SELECT id, name,
endpoint, vendor, cloud_instance_id,"
+ + " nameserver_id FROM rmq_instance")) {
+ while (rows.next()) {
+ ids.add(rows.getLong(1));
+ names.add(rows.getString(2));
+ endpoints.add(rows.getString(3));
+ vendors.add(rows.getString(4));
+ cloudIds.add(rows.getString(5));
+ nameserverIds.add(rows.getLong(6));
+ }
+ }
+ for (int i = 0; i < ids.size(); i++) {
+ String vendor = vendors.get(i);
+ if (StringUtils.hasText(vendor) && !"APACHE".equals(vendor)) {
+ if (!StringUtils.hasText(cloudIds.get(i))) {
+ throw new IllegalStateException("Cloud instance has no
cloud_instance_id and cannot be placed in a"
+ + " uniqueness domain; manual resolution required:
" + names.get(i));
}
+ domains.put(names.get(i),
ResourceOwnershipGuard.cloudDomain(cloudIds.get(i)));
+ continue;
+ }
+ if (nameserverIds.get(i) > 0) {
+ domains.put(names.get(i),
ResourceOwnershipGuard.nameserverDomain(nameserverIds.get(i)));
+ continue;
+ }
+ long nameserverId = requireNameserverId(connection, names.get(i),
endpoints.get(i));
+ try (PreparedStatement update =
connection.prepareStatement("UPDATE rmq_instance SET nameserver_id = ? WHERE id
= ?")) {
+ update.setLong(1, nameserverId);
+ update.setLong(2, ids.get(i));
+ update.executeUpdate();
}
+ domains.put(names.get(i),
ResourceOwnershipGuard.nameserverDomain(nameserverId));
}
+ return domains;
}
- private void ensureIndex(Connection connection, String table, String name,
String column, boolean unique)
+ private long requireNameserverId(Connection connection, String
instanceName, String endpoint) throws SQLException {
+ String normalized = NamesrvAddrParser.normalize(endpoint);
+ try (PreparedStatement select = connection.prepareStatement("SELECT id
FROM rmq_nameserver WHERE namesrv_addr = ?")) {
+ select.setString(1, normalized);
+ try (ResultSet rows = select.executeQuery()) {
+ if (rows.next()) {
+ return rows.getLong(1);
+ }
+ }
+ }
+ String name = "ns-" + instanceName;
+ try (PreparedStatement existing = connection.prepareStatement("SELECT
COUNT(*) FROM rmq_nameserver WHERE name = ?")) {
+ existing.setString(1, name);
+ try (ResultSet rows = existing.executeQuery()) {
+ rows.next();
+ if (rows.getLong(1) > 0) {
+ name = name + "-" + System.nanoTime();
+ }
+ }
+ }
+ try (PreparedStatement insert = connection.prepareStatement(
+ "INSERT INTO rmq_nameserver (name, namesrv_addr, status)
VALUES (?, ?, 'healthy')",
+ Statement.RETURN_GENERATED_KEYS)) {
+ insert.setString(1, name);
+ insert.setString(2, normalized);
+ insert.executeUpdate();
+ try (ResultSet keys = insert.getGeneratedKeys()) {
+ if (!keys.next()) {
+ throw new IllegalStateException("NameServer registration
returned no id for instance: " + instanceName);
+ }
+ return keys.getLong(1);
+ }
+ }
+ }
+
+ private void backfillResources(Connection connection, String table,
Map<String, String> domains) throws SQLException {
+ List<Long> ids = new ArrayList<>();
+ List<String> owners = new ArrayList<>();
+ try (Statement statement = connection.createStatement();
+ ResultSet rows = statement.executeQuery("SELECT id, instance_id,
cluster_id FROM " + table
+ + " WHERE domain = '' OR domain IS NULL")) {
+ while (rows.next()) {
+ ids.add(rows.getLong(1));
+ // Legacy unclaimed rows fall back to their cluster, which
used to double as the virtual instance name.
+ String instanceId = rows.getString(2);
+ owners.add(StringUtils.hasText(instanceId) ? instanceId :
rows.getString(3));
+ }
+ }
+ for (int i = 0; i < ids.size(); i++) {
+ String domain = domains.get(owners.get(i));
+ if (domain == null) {
+ throw new IllegalStateException("Resource ownership cannot be
determined and requires manual confirmation: "
+ + table + "/" + ids.get(i) + " owner=" +
owners.get(i));
+ }
+ try (PreparedStatement update =
connection.prepareStatement("UPDATE " + table + " SET domain = ? WHERE id =
?")) {
+ update.setString(1, domain);
+ update.setLong(2, ids.get(i));
+ update.executeUpdate();
+ }
+ }
+ }
+
+ private void ensureIndex(Connection connection, String table, String name,
List<String> columns, boolean unique)
throws SQLException {
Map<String, List<String>> indexes = indexes(connection, table);
- String expected = (unique ? "unique:" : "index:") + column;
+ List<String> expected = columns.stream().map(column -> (unique ?
"unique:" : "index:") + column).toList();
if (indexes.containsKey(name)) {
- if (!indexes.get(name).equals(List.of(expected))) {
+ if (!indexes.get(name).equals(expected)) {
throw new IllegalStateException("Index definition violates the
ownership constraint: " + table + "." + name);
}
return;
}
try (Statement statement = connection.createStatement()) {
statement.executeUpdate("CREATE " + (unique ? "UNIQUE " : "") +
"INDEX " + name
- + " ON " + table + " (" + column + ")");
+ + " ON " + table + " (" + String.join(", ", columns) +
")");
} catch (SQLException failure) {
// Concurrent startups only accept an identical index; failures
caused by duplicate data must not be swallowed.
- if (!List.of(expected).equals(indexes(connection,
table).get(name))) {
+ if (!expected.equals(indexes(connection, table).get(name))) {
+ throw failure;
+ }
+ }
+ }
+
+ private void dropIndex(Connection connection, String table, String name)
throws SQLException {
+ if (!indexes(connection, table).containsKey(name)) {
+ return;
+ }
+ try (Statement statement = connection.createStatement()) {
+ statement.executeUpdate("ALTER TABLE " + table + " DROP INDEX " +
name);
+ } catch (SQLException failure) {
+ if (indexes(connection, table).containsKey(name)) {
throw failure;
}
}
@@ -167,9 +286,9 @@ public class ResourceOwnershipSchemaMigration implements
InitializingBean {
String name = rows.getString("INDEX_NAME");
String column = rows.getString("COLUMN_NAME");
if (name != null && column != null) {
-
indexes.computeIfAbsent(name.toLowerCase(java.util.Locale.ROOT), ignored -> new
ArrayList<>())
+ indexes.computeIfAbsent(name.toLowerCase(Locale.ROOT),
ignored -> new ArrayList<>())
.add((rows.getBoolean("NON_UNIQUE") ? "index:" :
"unique:")
- +
column.toLowerCase(java.util.Locale.ROOT));
+ + column.toLowerCase(Locale.ROOT));
}
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/UpdateInstanceDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/UpdateInstanceDTO.java
index 63f48847b..06b38962f 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/UpdateInstanceDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/UpdateInstanceDTO.java
@@ -36,6 +36,8 @@ public class UpdateInstanceDTO {
private String adminCredentialRef;
+ private Long nameserverId;
+
public InstanceVO toInstanceVO() {
return InstanceVO.builder()
.name(name)
@@ -43,6 +45,7 @@ public class UpdateInstanceDTO {
.endpoint(endpoint)
.remark(remark)
.adminCredentialRef(adminCredentialRef)
+ .nameserverId(nameserverId)
.build();
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
index d5f8b7267..91f3f4aa5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
@@ -176,7 +176,7 @@ public class MessageService {
List<Resource> resources = List.of(new Resource(Kind.GROUP,
request.getConsumerGroup()),
ownershipGuard.topicResource(request.getTopic()));
if (apache) {
- resources.forEach(resource -> ownershipGuard.check(instance,
resource, true));
+ resources.forEach(resource ->
ownershipGuard.checkOwnerInDomain(instance, resource, true));
ownershipGuard.requireSupportedProvider(instance);
}
String detail = "topic=" + request.getTopic() + ", consumerGroup=" +
request.getConsumerGroup()
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicService.java
index faaeccdf9..4fed84f50 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicService.java
@@ -71,7 +71,7 @@ public class LiteTopicService {
var resource = ownershipGuard.topicResource(topicPattern);
// Ownership guard applies to open-source Apache instances only; cloud
calls are isolated by their own instance id.
if (instance.getVendor() == null || instance.getVendor() ==
InstanceVendor.APACHE) {
- ownershipGuard.check(instance, resource, true);
+ ownershipGuard.checkSlot(instance, resource, null, true);
ownershipGuard.requireSupportedProvider(instance);
}
if (resource.kind() != ResourceOwnershipGuard.Kind.TOPIC) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
index 64bb7720e..039a45d5c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
@@ -353,7 +353,7 @@ public class MetadataService {
if (retrySourceTopic != null &&
!retrySourceTopic.equals(original.getTopic())) {
resources.add(ownershipGuard.topicResource(retrySourceTopic));
}
- resources.forEach(resource -> ownershipGuard.check(instance, resource,
true));
+ resources.forEach(resource ->
ownershipGuard.checkOwnerInDomain(instance, resource, true));
return ownershipGuard.withOwned(instance, resources, () ->
sendMessage(request, retrySourceTopic));
}
@@ -754,7 +754,9 @@ public class MetadataService {
InstanceVO instance = ownershipGuard.requireInstance(instanceId);
// Ownership guard applies to open-source Apache instances only; cloud
calls are isolated by their own instance id.
if (instance.getVendor() == null || instance.getVendor() ==
InstanceVendor.APACHE) {
- ownershipGuard.check(instance, resource, required);
+ if (required) {
+ ownershipGuard.checkOwnerInDomain(instance, resource, true);
+ }
ownershipGuard.requireSupportedProvider(instance);
}
return instance.getName();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqGroup.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqGroup.java
index 959987243..879949991 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqGroup.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqGroup.java
@@ -34,6 +34,8 @@ public class RmqGroup {
private String instanceId;
+ private String domain;
+
private String name;
private String consumeType;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
index 11d7715f5..127418624 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
@@ -48,6 +48,8 @@ public class RmqInstance {
private String regionId;
+ private Long nameserverId;
+
private LocalDateTime gmtCreate;
private LocalDateTime gmtModified;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqTopic.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqTopic.java
index d2ec34350..7585ef2bf 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqTopic.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqTopic.java
@@ -34,6 +34,8 @@ public class RmqTopic {
private String instanceId;
+ private String domain;
+
private String name;
private String topicType;
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 1c718170f..3204b1677 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
@@ -280,7 +280,9 @@ public class RocketMQAdminClientImpl implements AdminClient
{
}
private TopicVO createTopic(String instanceId, TopicVO topic, boolean
importing) {
- String target = ownershipGuard.requireInstance(instanceId).getName();
+ InstanceVO instance = ownershipGuard.requireInstance(instanceId);
+ String target = instance.getName();
+ String domain = ownershipGuard.domainOf(instance);
if (topic == null) {
throw new BusinessException(400, "Topic request is required");
}
@@ -290,9 +292,12 @@ public class RocketMQAdminClientImpl implements
AdminClient {
int writeQueues = topic.getWriteQueues() > 0 ? topic.getWriteQueues()
: 8;
int readQueues = topic.getReadQueues() > 0 ? topic.getReadQueues() : 8;
- return executeResourceWrite(target, Kind.TOPIC, topicName, true,
topic.getType(), (admin, clusterName) -> {
+ var write = new WriteRequest(true, topic.getType(),
topic.getClusterId());
+ return executeResourceWrite(target, Kind.TOPIC, topicName, write,
(admin, clusterName) -> {
try {
RmqTopic existing = topicMapper.selectOne(new
LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getDomain, domain)
+ .eq(RmqTopic::getClusterId, clusterName)
.eq(RmqTopic::getName, topicName));
TopicPerm effectivePerm = topic.getPerm() != null
? topic.getPerm()
@@ -318,12 +323,15 @@ public class RocketMQAdminClientImpl implements
AdminClient {
// The global reservation was already committed independently
before the RPC; here we only update the unique record.
RmqTopic entity = topicMapper.selectOne(new
LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getDomain, domain)
+ .eq(RmqTopic::getClusterId, clusterName)
.eq(RmqTopic::getName, topicName));
boolean isNew = entity == null;
if (isNew) {
entity = new RmqTopic();
entity.setName(topicName);
entity.setClusterId(clusterName);
+ entity.setDomain(domain);
entity.setInstanceId(target);
entity.setGmtCreate(LocalDateTime.now());
}
@@ -365,14 +373,19 @@ public class RocketMQAdminClientImpl implements
AdminClient {
@Override
public TopicVO updateTopic(String instanceId, TopicVO topic) {
- String target = ownershipGuard.requireInstance(instanceId).getName();
+ InstanceVO instance = ownershipGuard.requireInstance(instanceId);
+ String target = instance.getName();
+ String domain = ownershipGuard.domainOf(instance);
topic.setInstanceId(target);
topic.setName(ResourceOwnershipGuard.requireText(topic.getName(),
"topicName"));
String topicName = topic.getName();
- return executeResourceWrite(target, Kind.TOPIC, topicName, false,
topic.getType(), (admin, clusterName) -> {
+ var write = new WriteRequest(false, topic.getType(),
topic.getClusterId());
+ return executeResourceWrite(target, Kind.TOPIC, topicName, write,
(admin, clusterName) -> {
try {
RmqTopic existing = topicMapper.selectOne(new
LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getDomain, domain)
+ .eq(RmqTopic::getClusterId, clusterName)
.eq(RmqTopic::getName, topicName));
if (existing == null) {
throw new BusinessException(404, "Topic not found: " +
topicName);
@@ -492,21 +505,26 @@ public class RocketMQAdminClientImpl implements
AdminClient {
Resource resource = ownershipGuard.topicResource(name);
if (resource.kind() != Kind.TOPIC) {
ownershipGuard.withOwned(instance, List.of(resource), () ->
executeForInstance(instance.getName(), admin -> {
- var owner = ownershipGuard.check(instance, resource, true);
+ var owner = ownershipGuard.checkOwnerInDomain(instance,
resource, true);
var target = ApacheWriteTargetResolver.resolve(admin,
instance, owner.clusterId());
deletePhysicalTopic(admin, instance.getName(), name,
target.cluster(), target.masters());
return null;
}));
return;
}
- executeResourceWrite(instance.getName(), Kind.TOPIC, name, false,
null, (admin, clusterName) -> {
+ ownershipGuard.checkOwnerInDomain(instance, resource, true);
+ ownershipGuard.withOwned(instance, List.of(resource), () ->
executeForInstance(instance.getName(), admin -> {
+ List<ResourceOwnershipGuard.Ownership> slots =
ownershipGuard.slots(instance, resource);
try {
- Set<String> brokerAddrs =
getMasterBrokerAddrsForCluster(admin, clusterName);
-
- deletePhysicalTopic(admin, instance.getName(), name,
clusterName, brokerAddrs);
- topicMapper.delete(new
LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName, name));
+ for (ResourceOwnershipGuard.Ownership slot : slots) {
+ deletePhysicalTopic(admin, instance.getName(), name,
slot.clusterId(),
+ getMasterBrokerAddrsForCluster(admin,
slot.clusterId()));
+ }
+ topicMapper.delete(new LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getDomain,
ownershipGuard.domainOf(instance))
+ .eq(RmqTopic::getName, name));
- recordAudit("DELETE_TOPIC", name, "", "SUCCESS");
+ recordAudit("DELETE_TOPIC", name, "slots=" + slots.size(),
"SUCCESS");
return null;
} catch (BusinessException e) {
recordAudit("DELETE_TOPIC", name, e.getMessage(), "FAILED");
@@ -515,7 +533,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
recordAudit("DELETE_TOPIC", name, e.getMessage(), "FAILED");
throw classifyBrokerFailure(e, "delete topic");
}
- });
+ }));
}
@Override
@@ -539,10 +557,10 @@ public class RocketMQAdminClientImpl implements
AdminClient {
request.setInstanceId(instance.getName());
request.setTopic(ResourceOwnershipGuard.requireText(request.getTopic(),
"topicName"));
Resource resource = ownershipGuard.topicResource(request.getTopic());
- ownershipGuard.check(instance, resource, true);
+ ownershipGuard.checkOwnerInDomain(instance, resource, true);
return ownershipGuard.withOwned(instance, List.of(resource), () -> {
executeForInstance(instance.getName(), admin -> {
- var owner = ownershipGuard.check(instance, resource, true);
+ var owner = ownershipGuard.checkOwnerInDomain(instance,
resource, true);
var target = ApacheWriteTargetResolver.resolve(admin,
instance, owner.clusterId());
ApacheWriteTargetResolver.requireTopicRoute(admin, target,
request.getTopic());
return null;
@@ -675,19 +693,25 @@ public class RocketMQAdminClientImpl implements
AdminClient {
}
private ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group, boolean
importing) {
- String instanceId = ownershipGuard.requireInstance(group == null ?
null : group.getInstanceId()).getName();
+ InstanceVO instance = ownershipGuard.requireInstance(group == null ?
null : group.getInstanceId());
+ String instanceId = instance.getName();
+ String domain = ownershipGuard.domainOf(instance);
group.setInstanceId(instanceId);
group.setName(ResourceOwnershipGuard.requireText(group.getName(),
"groupName"));
- return executeResourceWrite(instanceId, Kind.GROUP, group.getName(),
true, null,
- (admin, cluster) -> createConsumerGroup(admin, group, cluster,
importing));
+ var write = new WriteRequest(true, null, group.getClusterId());
+ return executeResourceWrite(instanceId, Kind.GROUP, group.getName(),
write,
+ (admin, cluster) -> createConsumerGroup(admin, group, cluster,
domain, importing));
}
@Override
public ConsumerGroupVO updateConsumerGroup(ConsumerGroupVO group) {
- String instanceId = ownershipGuard.requireInstance(group == null ?
null : group.getInstanceId()).getName();
+ InstanceVO instance = ownershipGuard.requireInstance(group == null ?
null : group.getInstanceId());
+ String instanceId = instance.getName();
+ String domain = ownershipGuard.domainOf(instance);
group.setInstanceId(instanceId);
group.setName(ResourceOwnershipGuard.requireText(group.getName(),
"groupName"));
- return executeResourceWrite(instanceId, Kind.GROUP, group.getName(),
false, null, (admin, clusterName) -> {
+ var write = new WriteRequest(false, null, group.getClusterId());
+ return executeResourceWrite(instanceId, Kind.GROUP, group.getName(),
write, (admin, clusterName) -> {
String groupName = group.getName();
int totalBrokers = 0;
int updatedBrokers = 0;
@@ -715,7 +739,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
updatedBrokers++;
}
- persistConsumerGroup(group, clusterName,
group.getRetryMaxTimes());
+ persistConsumerGroup(group, clusterName, domain,
group.getRetryMaxTimes());
recordAudit("UPDATE_GROUP", groupName, "retryMaxTimes=" +
group.getRetryMaxTimes()
+ ", brokersUpdated=" + updatedBrokers + "/" +
totalBrokers, "SUCCESS");
return group;
@@ -736,7 +760,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
return executeForInstance(instanceId, admin -> {
try {
InstanceVO instance =
ownershipGuard.requireInstance(instanceId);
- var owner = ownershipGuard.check(instance, new
Resource(Kind.GROUP, name), true);
+ var owner = ownershipGuard.checkSlot(instance, new
Resource(Kind.GROUP, name), null, true);
Set<String> brokerAddrs =
ApacheWriteTargetResolver.resolve(admin, instance, owner.clusterId()).masters();
if (brokerAddrs.isEmpty()) {
throw new BusinessException(502, "No broker available to
get consumer group settings");
@@ -761,7 +785,8 @@ public class RocketMQAdminClientImpl implements AdminClient
{
public ConsumerGroupSettingsVO updateConsumerGroupSettings(String
instanceId, String rawName,
ConsumerGroupSettingsCommand command) {
String name = ResourceOwnershipGuard.requireText(rawName, "groupName");
- return executeResourceWrite(instanceId, Kind.GROUP, name, false, null,
(admin, clusterName) -> {
+ var write = new WriteRequest(false, null, null);
+ return executeResourceWrite(instanceId, Kind.GROUP, name, write,
(admin, clusterName) -> {
int totalBrokers = 0;
int updatedBrokers = 0;
try {
@@ -836,7 +861,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
}
private ConsumerGroupVO createConsumerGroup(MQAdminExt admin,
ConsumerGroupVO group, String groupClusterName,
- boolean importing) {
+ String domain, boolean
importing) {
String groupName = group.getName();
try {
@@ -865,7 +890,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
}
}
- persistConsumerGroup(group, groupClusterName,
config.getRetryMaxTimes());
+ persistConsumerGroup(group, groupClusterName, domain,
config.getRetryMaxTimes());
recordAudit("CREATE_GROUP", groupName,
"retryMaxTimes=" + config.getRetryMaxTimes()
+ ", consumeMessageOrderly=" +
consumeMessageOrderly, "SUCCESS");
@@ -893,15 +918,18 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return normalized.contains("FIFO") || normalized.contains("ORDER");
}
- private void persistConsumerGroup(ConsumerGroupVO group, String
clusterName, int retryMaxTimes) {
- // Ownership is determined by the global reservation guard; a
three-column filter must not mask cross-instance conflicts.
+ private void persistConsumerGroup(ConsumerGroupVO group, String
clusterName, String domain, int retryMaxTimes) {
+ // Ownership is determined by the slot reservation guard; the slot key
must not mask cross-instance conflicts.
RmqGroup entity = groupMapper.selectOne(new
LambdaQueryWrapper<RmqGroup>()
+ .eq(RmqGroup::getDomain, domain)
+ .eq(RmqGroup::getClusterId, clusterName)
.eq(RmqGroup::getName, group.getName()));
boolean isNewGroup = entity == null;
if (isNewGroup) {
entity = new RmqGroup();
entity.setName(group.getName());
entity.setClusterId(clusterName);
+ entity.setDomain(domain);
entity.setInstanceId(group.getInstanceId());
entity.setGmtCreate(LocalDateTime.now());
}
@@ -920,15 +948,21 @@ public class RocketMQAdminClientImpl implements
AdminClient {
@Override
public void deleteConsumerGroup(String instanceId, String rawName) {
- String target = ownershipGuard.requireInstance(instanceId).getName();
+ InstanceVO instance = ownershipGuard.requireInstance(instanceId);
+ String target = instance.getName();
String name = ResourceOwnershipGuard.requireText(rawName, "groupName");
- executeResourceWrite(target, Kind.GROUP, name, false, null, (admin,
clusterName) -> {
- doDeleteConsumerGroup(target, admin, name, clusterName);
+ Resource resource = new Resource(Kind.GROUP, name);
+ ownershipGuard.checkOwnerInDomain(instance, resource, true);
+ ownershipGuard.withOwned(instance, List.of(resource), () ->
executeForInstance(target, admin -> {
+ for (ResourceOwnershipGuard.Ownership slot :
ownershipGuard.slots(instance, resource)) {
+ doDeleteConsumerGroup(target, admin, name, slot.clusterId(),
slot.domain());
+ }
return null;
- });
+ }));
}
- private void doDeleteConsumerGroup(String instanceId, MQAdminExt admin,
String name, String clusterName) {
+ private void doDeleteConsumerGroup(String instanceId, MQAdminExt admin,
String name, String clusterName,
+ String domain) {
try {
Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin,
clusterName);
@@ -940,7 +974,10 @@ public class RocketMQAdminClientImpl implements
AdminClient {
// Delete derived resources while holding group ownership; on
failure keep the reservation so the original instance can retry.
deletePhysicalTopic(admin, instanceId, "%DLQ%" + name,
clusterName, brokerAddrs);
deletePhysicalTopic(admin, instanceId, "%RETRY%" + name,
clusterName, brokerAddrs);
- groupMapper.delete(new
LambdaQueryWrapper<RmqGroup>().eq(RmqGroup::getName, name));
+ groupMapper.delete(new LambdaQueryWrapper<RmqGroup>()
+ .eq(RmqGroup::getDomain, domain)
+ .eq(RmqGroup::getClusterId, clusterName)
+ .eq(RmqGroup::getName, name));
recordAudit("DELETE_GROUP", name, "", "SUCCESS");
} catch (BusinessException e) {
@@ -1209,12 +1246,12 @@ public class RocketMQAdminClientImpl implements
AdminClient {
InstanceVO instance = ownershipGuard.requireInstance(instanceId);
Resource group = new Resource(Kind.GROUP,
ResourceOwnershipGuard.requireText(name, "groupName"));
Resource topicResource = ownershipGuard.topicResource(topic);
- ownershipGuard.check(instance, group, true);
- ownershipGuard.check(instance, topicResource, true);
+ ownershipGuard.checkOwnerInDomain(instance, group, true);
+ ownershipGuard.checkOwnerInDomain(instance, topicResource, true);
ownershipGuard.withOwned(instance, List.of(group, topicResource),
() ->
executeForInstance(instance.getName(), admin -> {
- var groupOwner = ownershipGuard.check(instance, group,
true);
- var topicOwner = ownershipGuard.check(instance,
topicResource, true);
+ var groupOwner =
ownershipGuard.checkOwnerInDomain(instance, group, true);
+ var topicOwner =
ownershipGuard.checkOwnerInDomain(instance, topicResource, true);
if
(!groupOwner.clusterId().equals(topicOwner.clusterId())) {
throw new BusinessException(409, "Group and topic
are not in the same target cluster");
}
@@ -1292,22 +1329,29 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return
runtimeAdminClientResolver.execute(ResourceOwnershipGuard.requireText(instanceId,
"instanceId"), action);
}
- private <T> T executeResourceWrite(String instanceId, Kind kind, String
name, boolean create, TopicType type,
+ private <T> T executeResourceWrite(String instanceId, Kind kind, String
name, WriteRequest request,
java.util.function.BiFunction<MQAdminExt, String, T> action) {
InstanceVO instance = ownershipGuard.requireInstance(instanceId);
Resource resource = new Resource(kind,
ResourceOwnershipGuard.requireText(name, "resourceName"));
- if (create && kind == Kind.TOPIC &&
ownershipGuard.topicResource(name).kind() != Kind.TOPIC) {
+ if (request.create() && kind == Kind.TOPIC &&
ownershipGuard.topicResource(name).kind() != Kind.TOPIC) {
throw new BusinessException(400, "Derived resources cannot be
registered as a regular topic");
}
- var owner = ownershipGuard.check(instance, resource, !create);
+ var owner = request.create()
+ ? ownershipGuard.createTarget(instance, resource)
+ : ownershipGuard.checkSlot(instance, resource, null, true);
ownershipGuard.requireSupportedProvider(instance);
return executeForInstance(instance.getName(), admin -> {
- var target = ApacheWriteTargetResolver.resolve(admin, instance,
owner == null ? null : owner.clusterId());
- return ownershipGuard.write(instance, resource, target.cluster(),
create,
- type == null ? null : type.name(), () ->
action.apply(admin, target.cluster()));
+ var target = ApacheWriteTargetResolver.resolve(admin, instance,
+ owner == null ? request.cluster() : owner.clusterId());
+ return ownershipGuard.write(instance, resource, target.cluster(),
request.create(),
+ request.type() == null ? null : request.type().name(),
+ () -> action.apply(admin, target.cluster()));
});
}
+ /** What a registration write intends: create or update, the message type,
and an explicit physical cluster. */
+ private record WriteRequest(boolean create, TopicType type, String
cluster) { }
+
/** Import reads all target masters within the ownership row lock;
existing config is only registered, never overwritten. */
private boolean verifyTopicImport(MQAdminExt admin, Set<String> masters,
TopicConfig requested) throws Exception {
boolean found = false;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
index 4211ccd6f..06ccfb960 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
@@ -370,7 +370,7 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
var instance = ownershipGuard.requireInstance(instanceId);
String topicPattern =
ResourceOwnershipGuard.requireText(rawTopicPattern, "topicPattern");
var resource = ownershipGuard.topicResource(topicPattern);
- ownershipGuard.check(instance, resource, true);
+ ownershipGuard.checkSlot(instance, resource, null, true);
ownershipGuard.requireSupportedProvider(instance);
if (resource.kind() != ResourceOwnershipGuard.Kind.TOPIC) {
throw new BusinessException(409, "TTL can only be updated for a
registered Lite parent topic");
@@ -380,7 +380,7 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
}
long minutes = Math.min(Math.max(Math.round(ttlMillis / 60000.0), 1),
MAX_LITE_TTL_MINUTES);
ownershipGuard.withOwned(instance, List.of(resource), () ->
runtimeAdminClientResolver.execute(instance, admin -> {
- var owner = ownershipGuard.check(instance, resource, true);
+ var owner = ownershipGuard.checkSlot(instance, resource, null,
true);
var target = ApacheWriteTargetResolver.resolve(admin, instance,
owner.clusterId());
// Read every master's config before writing any of them: a master
that cannot be
// examined must fail the request up front instead of being
silently skipped, which
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
index 4afed6fb9..4a4116fb9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
@@ -558,12 +558,12 @@ public class RocketMQMessageProvider implements
MessageProvider {
String messageId =
ResourceOwnershipGuard.requireText(request.getMsgId(), "msgId");
Resource topicResource = ownershipGuard.topicResource(topic);
Resource groupResource = new Resource(Kind.GROUP, group);
- ownershipGuard.check(instance, topicResource, true);
- ownershipGuard.check(instance, groupResource, true);
+ ownershipGuard.checkOwnerInDomain(instance, topicResource, true);
+ ownershipGuard.checkOwnerInDomain(instance, groupResource, true);
return ownershipGuard.withOwned(instance, List.of(topicResource,
groupResource), () ->
runtimeAdminClientResolver.execute(instance.getName(), admin
-> {
- var topicOwner = ownershipGuard.check(instance,
topicResource, true);
- var groupOwner = ownershipGuard.check(instance,
groupResource, true);
+ var topicOwner =
ownershipGuard.checkOwnerInDomain(instance, topicResource, true);
+ var groupOwner =
ownershipGuard.checkOwnerInDomain(instance, groupResource, true);
if
(!topicOwner.clusterId().equals(groupOwner.clusterId())) {
throw new BusinessException(409, "Group and topic are
not in the same target cluster");
}
diff --git a/server/src/main/resources/db/schema.sql
b/server/src/main/resources/db/schema.sql
index 094468060..c44222b0a 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -101,9 +101,11 @@ CREATE TABLE IF NOT EXISTS rmq_instance (
credential_id bigint(20) unsigned COMMENT '引用 rmq_cloud_credential.id',
admin_credential_ref VARCHAR(128) COMMENT 'External Apache admin credential
reference; no secret material',
region_id VARCHAR(128),
+ nameserver_id bigint(20) unsigned COMMENT '绑定的
rmq_nameserver.id；应用层校验存在性，不加外键；APACHE 实例创建时必填且不可修改',
PRIMARY KEY (`id`),
UNIQUE KEY uk_instance_name (name),
INDEX idx_instance_credential_id (credential_id),
+ INDEX idx_instance_nameserver_id (nameserver_id),
CONSTRAINT fk_instance_cloud_credential FOREIGN KEY (credential_id)
REFERENCES rmq_cloud_credential(id) ON DELETE RESTRICT
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
@@ -125,6 +127,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_topic (
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT '修改时间',
cluster_id VARCHAR(64) NOT NULL,
instance_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '归属实例
ID（rmq_instance.name；空字符串表示历史未归属记录）',
+ domain VARCHAR(128) NOT NULL DEFAULT '' COMMENT '唯一性域：ns:<nameserver_id> 或
cloud:<cloud_instance_id>',
name VARCHAR(255) NOT NULL,
topic_type VARCHAR(32) DEFAULT 'NORMAL',
read_queue_nums INT DEFAULT 8,
@@ -134,9 +137,10 @@ CREATE TABLE IF NOT EXISTS rmq_instance_topic (
status VARCHAR(32) DEFAULT 'ACTIVE',
created_by VARCHAR(64),
PRIMARY KEY (`id`),
- UNIQUE KEY uk_topic_name (name),
+ UNIQUE KEY uk_topic_domain (domain, cluster_id, name),
INDEX idx_topic_instance (instance_id),
- INDEX idx_topic_cluster (cluster_id)
+ INDEX idx_topic_cluster (cluster_id),
+ INDEX idx_topic_domain (domain)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 4. Consumer Group 管理记录
@@ -146,6 +150,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_group (
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT '修改时间',
cluster_id VARCHAR(64) NOT NULL,
instance_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '归属实例
ID（rmq_instance.name；空字符串表示历史未归属记录）',
+ domain VARCHAR(128) NOT NULL DEFAULT '' COMMENT '唯一性域：ns:<nameserver_id> 或
cloud:<cloud_instance_id>',
name VARCHAR(255) NOT NULL,
consume_type VARCHAR(32) DEFAULT 'CONCURRENTLY',
message_model VARCHAR(32) DEFAULT 'CLUSTERING',
@@ -153,9 +158,10 @@ CREATE TABLE IF NOT EXISTS rmq_instance_group (
status VARCHAR(32) DEFAULT 'ACTIVE',
created_by VARCHAR(64),
PRIMARY KEY (`id`),
- UNIQUE KEY uk_group_name (name),
+ UNIQUE KEY uk_group_domain (domain, cluster_id, name),
INDEX idx_group_instance (instance_id),
- INDEX idx_group_cluster (cluster_id)
+ INDEX idx_group_cluster (cluster_id),
+ INDEX idx_group_domain (domain)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 5. K8s 证书管理
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
index efbddcf2c..a5182abe2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
@@ -109,6 +109,9 @@ class InstanceServiceTest {
@Mock
private ResourceOwnershipGuard ownershipGuard;
+ @Mock
+ private
org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryService
nameserverRegistry;
+
@InjectMocks
private InstanceService instanceService;
@@ -133,7 +136,7 @@ class InstanceServiceTest {
@Test
void updateOwnedInstanceStopsBeforeSaveTest() {
InstanceVO existing = InstanceVO.builder().name("instance-a")
- .endpoint("old:9876").vendor(InstanceVendor.APACHE).build();
+
.endpoint("old:9876").vendor(InstanceVendor.APACHE).nameserverId(1L).build();
existing.setId(1L);
InstanceVO request = InstanceVO.builder().endpoint("new:9876").build();
request.setId(1L);
@@ -483,7 +486,7 @@ class InstanceServiceTest {
.name("new-instance")
.endpoint("10.0.1.1:8080")
.type(InstanceType.PROXY_CLUSTER)
- .build();
+ .nameserverId(1L).build();
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation -> {
InstanceVO saved = invocation.getArgument(0);
@@ -569,7 +572,7 @@ class InstanceServiceTest {
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
String atLimit = "n".repeat(512);
InstanceVO accepted =
InstanceVO.builder().name("inst-a").type(InstanceType.PROXY_CLUSTER)
- .endpoint(atLimit).build();
+ .endpoint(atLimit).nameserverId(1L).build();
assertThat(instanceService.createInstance(accepted).getEndpoint()).isEqualTo(atLimit);
@@ -607,7 +610,7 @@ class InstanceServiceTest {
// value fits the column and counting chars would reject it as a 500
from the write instead.
String atLimit =
EMOJI.repeat(InstanceService.MAX_INSTANCE_ENDPOINT_LENGTH);
InstanceVO accepted =
InstanceVO.builder().name("inst-a").type(InstanceType.PROXY_CLUSTER)
- .endpoint(atLimit).build();
+ .endpoint(atLimit).nameserverId(1L).build();
assertThat(instanceService.createInstance(accepted).getEndpoint()).isEqualTo(atLimit);
@@ -625,12 +628,12 @@ class InstanceServiceTest {
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
String atLimit = "r".repeat(255);
InstanceVO accepted =
InstanceVO.builder().name("inst-a").type(InstanceType.PROXY_CLUSTER)
- .endpoint("namesrv:9876").remark(atLimit).build();
+
.endpoint("namesrv:9876").remark(atLimit).nameserverId(1L).build();
assertThat(instanceService.createInstance(accepted).getRemark()).isEqualTo(atLimit);
InstanceVO rejected =
InstanceVO.builder().name("inst-b").type(InstanceType.PROXY_CLUSTER)
- .endpoint("namesrv:9876").remark(atLimit + "r").build();
+ .endpoint("namesrv:9876").remark(atLimit +
"r").nameserverId(1L).build();
assertThatThrownBy(() -> instanceService.createInstance(rejected))
.isInstanceOf(BusinessException.class)
.hasMessage("InstanceVO remark must not exceed 255
characters");
@@ -661,7 +664,7 @@ class InstanceServiceTest {
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
String atLimit = "c".repeat(128);
InstanceVO accepted =
InstanceVO.builder().name("inst-a").type(InstanceType.PROXY_CLUSTER)
- .endpoint("namesrv:9876").adminCredentialRef(atLimit).build();
+
.endpoint("namesrv:9876").adminCredentialRef(atLimit).nameserverId(1L).build();
assertThat(instanceService.createInstance(accepted).getAdminCredentialRef()).isEqualTo(atLimit);
@@ -693,7 +696,7 @@ class InstanceServiceTest {
@Test
void createInstanceShouldTrimEndpointBeforeSaving() {
InstanceVO input =
InstanceVO.builder().name("valid-name").type(InstanceType.PROXY_CLUSTER)
- .endpoint(" namesrv:9876 ").build();
+ .endpoint(" namesrv:9876 ").nameserverId(1L).build();
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
assertThat(instanceService.createInstance(input).getEndpoint()).isEqualTo("namesrv:9876");
@@ -702,7 +705,7 @@ class InstanceServiceTest {
@Test
void createInstanceShouldTrimNameBeforeSaving() {
InstanceVO input = InstanceVO.builder().name(" production
").type(InstanceType.PROXY_CLUSTER)
- .endpoint("namesrv:9876").build();
+ .endpoint("namesrv:9876").nameserverId(1L).build();
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
assertThat(instanceService.createInstance(input).getName()).isEqualTo("production");
@@ -711,7 +714,7 @@ class InstanceServiceTest {
@Test
void
createApacheInstanceShouldTrimAndPersistOnlyAdminCredentialReference() {
InstanceVO input =
InstanceVO.builder().name("production").type(InstanceType.PROXY_CLUSTER)
- .endpoint("namesrv:9876").adminCredentialRef("
production-admin ").build();
+ .endpoint("namesrv:9876").adminCredentialRef("
production-admin ").nameserverId(1L).build();
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
InstanceVO saved = instanceService.createInstance(input);
@@ -837,69 +840,49 @@ class InstanceServiceTest {
}
@Test
- void updateInstanceShouldReleaseUnusedPreviousEndpoint() {
+ void updateInstanceShouldRejectEndpointOutsideTheBoundNameserverTest() {
InstanceVO existing = InstanceVO.builder()
.name("instance")
+ .type(InstanceType.DIRECT)
.endpoint("old-namesrv:9876")
+ .nameserverId(1L)
.build();
existing.setId(1L);
InstanceVO update =
InstanceVO.builder().endpoint("new-namesrv:9876").build();
update.setId(1L);
when(instanceRepository.findById(1L)).thenReturn(Optional.of(existing));
-
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
- when(instanceRepository.findAll()).thenReturn(List.of());
-
- instanceService.updateInstance(update);
-
- verify(adminFactory).release("old-namesrv:9876");
- }
-
- @Test
- void
updateInstanceShouldKeepEndpointClientWhenAnotherInstanceReferencesIt() {
- InstanceVO existing = InstanceVO.builder()
- .name("instance")
- .endpoint("shared-namesrv:9876")
- .build();
- existing.setId(1L);
- InstanceVO shared = InstanceVO.builder()
- .name("shared-instance")
- .endpoint(" shared-namesrv:9876 ")
- .build();
- shared.setId(2L);
- InstanceVO update =
InstanceVO.builder().endpoint("new-namesrv:9876").build();
- update.setId(1L);
-
-
when(instanceRepository.findById(1L)).thenReturn(Optional.of(existing));
-
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
- when(instanceRepository.findAll()).thenReturn(List.of(shared));
-
- instanceService.updateInstance(update);
+ when(nameserverRegistry.addressOf(1L))
+
.thenReturn(org.apache.rocketmq.studio.cluster.nameserver.NamesrvAddrParser.normalize("old-namesrv:9876"));
- verify(adminFactory, never()).release("shared-namesrv:9876");
+ assertThatThrownBy(() -> instanceService.updateInstance(update))
+ .isInstanceOfSatisfying(BusinessException.class, ex ->
assertThat(ex.getCode()).isEqualTo(400));
+ verify(adminFactory, org.mockito.Mockito.never()).release(anyString());
+ verify(instanceRepository,
org.mockito.Mockito.never()).save(any(InstanceVO.class));
}
@Test
- void
updateInstanceShouldKeepEndpointClientWhenAnotherInstanceUsesEquivalentAddressList()
{
+ void
updateInstanceShouldAcceptAnEquivalentRewriteOfTheBoundAddressListTest() {
InstanceVO existing = InstanceVO.builder()
.name("instance")
- .endpoint("namesrv-b:9876;namesrv-a:9876")
+ .type(InstanceType.DIRECT)
+ .endpoint("namesrv-a:9876;namesrv-b:9876")
+ .nameserverId(1L)
.build();
existing.setId(1L);
- InstanceVO shared = InstanceVO.builder()
- .name("shared-instance")
- .endpoint(" namesrv-a:9876, namesrv-b:9876 ")
- .build();
- shared.setId(2L);
- InstanceVO update =
InstanceVO.builder().endpoint("new-namesrv:9876").build();
+ InstanceVO update = InstanceVO.builder().endpoint(" namesrv-a:9876,
namesrv-b:9876 ").build();
update.setId(1L);
when(instanceRepository.findById(1L)).thenReturn(Optional.of(existing));
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
- when(instanceRepository.findAll()).thenReturn(List.of(shared));
+
when(nameserverRegistry.addressOf(1L)).thenReturn(org.apache.rocketmq.studio.cluster.nameserver
+ .NamesrvAddrParser.normalize(" namesrv-a:9876, namesrv-b:9876
"));
- instanceService.updateInstance(update);
+ InstanceVO saved = instanceService.updateInstance(update);
+ assertThat(saved.getEndpoint())
+
.isEqualTo(org.apache.rocketmq.studio.cluster.nameserver.NamesrvAddrParser
+ .normalize(" namesrv-a:9876, namesrv-b:9876 "));
verifyNoInteractions(adminFactory);
}
@@ -1362,7 +1345,7 @@ class InstanceServiceTest {
.name("inst")
.endpoint("10.0.0.1:8080")
.type(InstanceType.PROXY_CLUSTER)
- .build();
+ .nameserverId(1L).build();
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
InstanceVO created = instanceService.createInstance(instance);
@@ -1378,7 +1361,7 @@ class InstanceServiceTest {
.name("local-proxy")
.endpoint("broker-proxy:8080")
.type(InstanceType.PROXY_LOCAL)
- .build();
+ .nameserverId(1L).build();
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
InstanceVO created = instanceService.createInstance(instance);
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 d5f21dba2..4755a19ad 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
@@ -58,6 +58,9 @@ class MybatisPlusInstanceRepositoryTest {
@Mock
private RmqGroupMapper groupMapper;
+ @Mock
+ private
org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryService
nameserverRegistry;
+
@InjectMocks
private MybatisPlusInstanceRepository repository;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
index 0f0407e58..5a8d5c57b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
@@ -79,6 +79,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
@@ -104,8 +105,9 @@ class ResourceOwnershipGuardTest {
resolver = mock(InstanceResolver.class);
defaultClusters = mock(RocketMQDefaultClusterResolver.class);
guard = new ResourceOwnershipGuard(jdbc, transactions, resolver);
- jdbc.update("INSERT INTO rmq_instance (name, type, endpoint, vendor)
VALUES ('first', 'DIRECT', 'same:9876', 'APACHE')");
- jdbc.update("INSERT INTO rmq_instance (name, type, endpoint, vendor)
VALUES ('second', 'DIRECT', 'same:9876', 'APACHE')");
+ jdbc.update("INSERT INTO rmq_nameserver (name, namesrv_addr, status)
VALUES ('ns-test', 'same:9876', 'healthy')");
+ jdbc.update("INSERT INTO rmq_instance (name, type, endpoint, vendor,
nameserver_id) VALUES ('first', 'DIRECT', 'same:9876', 'APACHE', 1)");
+ jdbc.update("INSERT INTO rmq_instance (name, type, endpoint, vendor,
nameserver_id) VALUES ('second', 'DIRECT', 'same:9876', 'APACHE', 1)");
first = instance("first");
second = instance("second");
}
@@ -184,7 +186,7 @@ class ResourceOwnershipGuardTest {
jdbc.update("DELETE FROM rmq_instance_topic WHERE name =
'orders'");
throw new BusinessException(502, "NameServer failed");
})).isInstanceOf(BusinessException.class);
- assertCode(() -> guard.check(second, topic, true), 409);
+ assertCode(() -> guard.checkSlot(second, topic, null, true), 409);
}
@Test
@@ -198,11 +200,50 @@ class ResourceOwnershipGuardTest {
}
@Test
- void legacyOwnerOnlyAcceptsConfirmedVirtualClusterTest() {
- jdbc.update("INSERT INTO rmq_instance_topic (name, cluster_id) VALUES
('orders', 'cluster')");
- assertCode(() -> guard.check(first, topic, true), 409);
- InstanceVO virtual =
InstanceVO.builder().name("cluster").vendor(InstanceVendor.APACHE).build();
- assertThat(guard.check(virtual, topic, true).instanceId()).isEmpty();
+ void sameNameInDifferentDomainsCoexistsTest() {
+ jdbc.update("INSERT INTO rmq_nameserver (name, namesrv_addr, status)
VALUES ('ns-other', 'other:9876', 'healthy')");
+ jdbc.update("INSERT INTO rmq_instance (name, type, endpoint, vendor,
nameserver_id)"
+ + " VALUES ('third', 'DIRECT', 'other:9876', 'APACHE', 2)");
+ InstanceVO third = instance("third");
+ guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
+ guard.write(third, topic, "cluster", true, "NORMAL", () -> null);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM
rmq_instance_topic", Integer.class)).isEqualTo(2);
+ }
+
+ @Test
+ void sameNameOnDifferentClustersOfOneDomainCoexistsTest() {
+ guard.write(first, topic, "cluster-a", true, "NORMAL", () -> null);
+ guard.write(second, topic, "cluster-b", true, "NORMAL", () -> null);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM
rmq_instance_topic", Integer.class)).isEqualTo(2);
+ assertThat(guard.slots(first, topic)).hasSize(2);
+ }
+
+ @Test
+ void sameSlotSecondInstanceIsRejectedTest() {
+ guard.write(first, topic, "cluster-a", true, "NORMAL", () -> null);
+ assertCode(() -> guard.write(second, topic, "cluster-a", true,
"NORMAL", () -> null), 409);
+ }
+
+ @Test
+ void sameSlotSameInstanceIsAnUpsertTest() {
+ guard.write(first, topic, "cluster-a", true, "NORMAL", () -> null);
+ guard.write(first, topic, "cluster-a", true, "NORMAL", () -> null);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM
rmq_instance_topic", Integer.class)).isEqualTo(1);
+ }
+
+ @Test
+ void releasedSlotCanBeClaimedByAnotherInstanceTest() {
+ guard.write(first, topic, "cluster-a", true, "NORMAL", () -> null);
+ jdbc.update("DELETE FROM rmq_instance_topic WHERE name = 'orders'");
+ guard.write(second, topic, "cluster-a", true, "NORMAL", () -> null);
+ assertThat(jdbc.queryForObject("SELECT instance_id FROM
rmq_instance_topic", String.class)).isEqualTo("second");
+ }
+
+ @Test
+ void foreignOwnerInSameDomainIsRejectedBeforeAnyWriteTest() {
+ jdbc.update("INSERT INTO rmq_instance_topic (name, cluster_id, domain,
instance_id) VALUES ('orders', 'cluster', 'ns:1', 'first')");
+ assertCode(() -> guard.checkSlot(second, topic, null, true), 409);
+ assertThat(guard.checkSlot(first, topic, null,
true).instanceId()).isEqualTo("first");
}
@Test
@@ -210,31 +251,35 @@ class ResourceOwnershipGuardTest {
legacyIndexes();
guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
jdbc.execute("DROP TABLE rmq_instance_ownership_lock");
- var migration = new ResourceOwnershipSchemaMigration(dataSource,
defaultClusters);
+ var migration = new ResourceOwnershipSchemaMigration(dataSource);
migration.afterPropertiesSet();
migration.afterPropertiesSet();
guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
- assertThatThrownBy(() -> jdbc.update("INSERT INTO rmq_instance_topic
(name, instance_id, cluster_id)"
- + " VALUES ('orders', 'second',
'other')")).isInstanceOf(org.springframework.dao.DuplicateKeyException.class);
+ // The slot key still rejects a second owner of the very same (domain,
cluster, name)...
+ assertThatThrownBy(() -> jdbc.update("INSERT INTO rmq_instance_topic
(name, instance_id, cluster_id, domain)"
+ + " VALUES ('orders', 'second', 'cluster',
'ns:1')")).isInstanceOf(org.springframework.dao.DuplicateKeyException.class);
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM
rmq_instance_topic", Integer.class)).isEqualTo(1);
+ // ...while the same name on another broker cluster of the domain is
exactly what the model allows.
+ jdbc.update("INSERT INTO rmq_instance_topic (name, instance_id,
cluster_id, domain)"
+ + " VALUES ('orders', 'second', 'other', 'ns:1')");
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM
rmq_instance_topic", Integer.class)).isEqualTo(2);
}
@Test
void migrationRejectsDuplicatesWithoutDiscardingAnyRowsTest() {
legacyIndexes();
- jdbc.update("INSERT INTO rmq_instance_group (name, instance_id,
cluster_id) VALUES ('buyers', 'first', 'a')");
- jdbc.update("INSERT INTO rmq_instance_group (name, instance_id,
cluster_id) VALUES ('buyers', 'second', 'b')");
- assertThatThrownBy(() -> new
ResourceOwnershipSchemaMigration(dataSource,
defaultClusters).afterPropertiesSet())
-
.isInstanceOf(IllegalStateException.class).hasMessageContaining("duplicate
name");
+ jdbc.update("INSERT INTO rmq_instance_group (name, instance_id,
cluster_id, domain) VALUES ('buyers', 'first', 'a', 'ns:1')");
+ jdbc.update("INSERT INTO rmq_instance_group (name, instance_id,
cluster_id, domain) VALUES ('buyers', 'second', 'a', 'ns:1')");
+ assertThatThrownBy(() -> new
ResourceOwnershipSchemaMigration(dataSource).afterPropertiesSet())
+
.isInstanceOf(IllegalStateException.class).hasMessageContaining("manual
resolution");
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM
rmq_instance_group", Integer.class)).isEqualTo(2);
}
@Test
void migrationRejectsUnknownLegacyOwnerWithoutChangingItTest() {
jdbc.update("INSERT INTO rmq_instance_topic (name, cluster_id) VALUES
('orders', 'unknown')");
- when(defaultClusters.names()).thenReturn(List.of("cluster"));
- assertThatThrownBy(() -> new
ResourceOwnershipSchemaMigration(dataSource,
defaultClusters).afterPropertiesSet())
-
.isInstanceOf(IllegalStateException.class).hasMessageContaining("no trusted
virtual cluster mapping");
+ assertThatThrownBy(() -> new
ResourceOwnershipSchemaMigration(dataSource).afterPropertiesSet())
+
.isInstanceOf(IllegalStateException.class).hasMessageContaining("cannot be
determined");
assertThat(jdbc.queryForObject("SELECT instance_id FROM
rmq_instance_topic", String.class)).isEmpty();
}
@@ -243,21 +288,23 @@ class ResourceOwnershipGuardTest {
var runtime = mock(RuntimeAdminClientResolver.class);
var broker = mock(DefaultMQAdminExt.class);
var admin = apacheAdmin(runtime, broker);
- InstanceVO virtual =
InstanceVO.builder().name("cluster-b").endpoint("same:9876").build();
-
when(resolver.findByIdentifier("cluster-b")).thenReturn(Optional.of(virtual));
+ jdbc.update("INSERT INTO rmq_instance (name, type, endpoint, vendor,
nameserver_id) VALUES ('prod-b', 'DIRECT', 'same:9876', 'APACHE', 1)");
+ InstanceVO registered = instance("prod-b");
+
when(resolver.findByIdentifier("prod-b")).thenReturn(Optional.of(registered));
when(broker.examineBrokerClusterInfo()).thenReturn(twoClusters());
TopicVO request = new TopicVO();
request.setName("orders");
- admin.createTopic("cluster-b", request);
+ request.setClusterId("cluster-b");
+ admin.createTopic("prod-b", request);
- assertThat(guard.check(virtual, topic,
true).clusterId()).isEqualTo("cluster-b");
+ assertThat(guard.checkSlot(registered, topic, null,
true).clusterId()).isEqualTo("cluster-b");
verify(broker).createAndUpdateTopicConfig(eq("10.0.1.1:10911"),
any(TopicConfig.class));
verify(broker,
never()).createAndUpdateTopicConfig(eq("10.0.0.1:10911"),
any(TopicConfig.class));
verify(broker,
never()).createAndUpdateTopicConfig(eq("10.0.1.2:10911"),
any(TopicConfig.class));
}
@Test
- void apacheWritesRejectForeignOwnerBeforeAnyBrokerOrProducerCallTest() {
+ void apacheWritesRejectForeignOwnerBeforeAnyBrokerOrProducerCallTest()
throws Exception {
guard.write(first, topic, "cluster-a", true, "NORMAL", () -> null);
guard.write(first, new Resource(Kind.GROUP, "buyers"), "cluster-a",
true, null, () -> null);
var runtime = mock(RuntimeAdminClientResolver.class);
@@ -285,7 +332,10 @@ class ResourceOwnershipGuardTest {
direct.setClientId("client");
direct.setMsgId("msg");
assertCode(() -> new RocketMQMessageProvider(runtime,
guard).consumeMessageDirectly(direct), 409);
- verifyNoInteractions(runtime, broker);
+ verify(broker, never()).createAndUpdateTopicConfig(anyString(), any());
+ verify(broker,
never()).createAndUpdateSubscriptionGroupConfig(anyString(), any());
+ verify(broker, never()).deleteTopicInBroker(any(), anyString());
+ verify(broker, never()).deleteSubscriptionGroup(anyString(),
anyString(), anyBoolean());
}
@Test
@@ -359,9 +409,8 @@ class ResourceOwnershipGuardTest {
@Test
void migrationRejectsMissingExplicitOwnerWithoutDataLossTest() {
jdbc.update("INSERT INTO rmq_instance_topic (name, instance_id,
cluster_id) VALUES ('orders', 'gone', 'cluster')");
- when(defaultClusters.names()).thenReturn(List.of("cluster"));
- assertThatThrownBy(() -> new
ResourceOwnershipSchemaMigration(dataSource,
defaultClusters).afterPropertiesSet())
-
.isInstanceOf(IllegalStateException.class).hasMessageContaining("does not exist
and has no trusted virtual mapping");
+ assertThatThrownBy(() -> new
ResourceOwnershipSchemaMigration(dataSource).afterPropertiesSet())
+
.isInstanceOf(IllegalStateException.class).hasMessageContaining("cannot be
determined");
assertThat(jdbc.queryForObject("SELECT instance_id FROM
rmq_instance_topic", String.class)).isEqualTo("gone");
}
@@ -381,7 +430,7 @@ class ResourceOwnershipGuardTest {
request.setName("orders");
request.setType(TopicType.NORMAL);
admin.importTopic(" first ", request);
- assertThat(guard.check(first, topic,
true).clusterId()).isEqualTo("cluster-b");
+ assertThat(guard.checkSlot(first, topic, null,
true).clusterId()).isEqualTo("cluster-b");
assertThat(jdbc.queryForObject("SELECT status FROM
rmq_instance_topic", String.class)).isEqualTo("ACTIVE");
request.setReadQueues(16);
assertCode(() -> admin.importTopic("first", request), 409);
@@ -408,7 +457,7 @@ class ResourceOwnershipGuardTest {
request.setName("orders");
request.setType(type);
admin.importTopic("first", request);
- assertThat(guard.check(first, topic,
true).topicType()).isEqualTo(type.name());
+ assertThat(guard.checkSlot(first, topic, null,
true).topicType()).isEqualTo(type.name());
physical.setAttributes(Map.of("message.type", type == TopicType.NORMAL
? "FIFO" : "NORMAL"));
assertCode(() -> admin.importTopic("first", request), 409);
verify(broker, never()).createAndUpdateTopicConfig(anyString(), any());
@@ -426,7 +475,7 @@ class ResourceOwnershipGuardTest {
topology.getBrokerAddrTable().remove("broker-a");
when(broker.examineBrokerClusterInfo()).thenReturn(topology);
org.mockito.Mockito.doAnswer(call -> {
- assertThat(guard.check(first, topic,
true).instanceId()).isEqualTo("first");
+ assertThat(guard.checkSlot(first, topic, null,
true).instanceId()).isEqualTo("first");
throw new BusinessException(502, "Remote already partially
succeeded");
}).when(broker).createAndUpdateTopicConfig(anyString(), any());
TopicVO request = new TopicVO();
@@ -493,29 +542,27 @@ class ResourceOwnershipGuardTest {
assertCode(() -> lite.extendTTL("first", "orders", 120_000L), 502);
assertCode(() -> lite.extendTTL("second", "orders", 120_000L), 409);
verifyNoInteractions(factory);
- assertThat(guard.check(first, topic,
true).instanceId()).isEqualTo("first");
+ assertThat(guard.checkSlot(first, topic, null,
true).instanceId()).isEqualTo("first");
}
@Test
- void virtualClaimAndRegistrationSerializeWithoutChangingOwnershipTest()
throws Exception {
- InstanceVO virtual = InstanceVO.builder().name("virtual").build();
+ void registeredClaimAndRegistrationSerializeWithoutChangingOwnershipTest()
throws Exception {
CountDownLatch entered = new CountDownLatch(1);
CountDownLatch release = new CountDownLatch(1);
try (var executor = Executors.newFixedThreadPool(2)) {
- var write = executor.submit(() -> guard.write(virtual, topic,
"virtual", true, "NORMAL", () -> {
+ var write = executor.submit(() -> guard.write(first, topic,
"cluster", true, "NORMAL", () -> {
entered.countDown();
await(release);
return null;
}));
assertThat(entered.await(5, TimeUnit.SECONDS)).isTrue();
- var register = executor.submit(() -> assertCode(() ->
guard.withInstanceRegistration("virtual", () ->
- jdbc.update("INSERT INTO rmq_instance (name, type,
endpoint) VALUES ('virtual','DIRECT','other:9876')")), 409));
+ var register = executor.submit(() -> assertCode(() ->
guard.withInstanceRegistration("first", () ->
+ jdbc.update("INSERT INTO rmq_instance (name, type,
endpoint, vendor, nameserver_id) VALUES ('dup', 'DIRECT', 'other:9876',
'APACHE', 1)")), 409));
release.countDown();
write.get(10, TimeUnit.SECONDS);
register.get(10, TimeUnit.SECONDS);
}
- assertThat(guard.check(virtual, topic,
true).instanceId()).isEqualTo("virtual");
- assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_instance
WHERE name = 'virtual'", Integer.class)).isZero();
+ assertThat(guard.checkSlot(first, topic, null,
true).instanceId()).isEqualTo("first");
}
@Test
@@ -539,13 +586,12 @@ class ResourceOwnershipGuardTest {
@Test
void migrationAcceptsTrustedVirtualOwnerWithoutRewritingLegacyDataTest()
throws Exception {
- jdbc.update("INSERT INTO rmq_instance_topic (name, cluster_id) VALUES
('orders', 'cluster')");
- when(defaultClusters.names()).thenReturn(List.of("cluster"));
- var migration = new ResourceOwnershipSchemaMigration(dataSource,
defaultClusters);
+ jdbc.update("INSERT INTO rmq_instance_topic (name, instance_id,
cluster_id) VALUES ('orders', 'first', 'cluster')");
+ var migration = new ResourceOwnershipSchemaMigration(dataSource);
migration.afterPropertiesSet();
migration.afterPropertiesSet();
- assertThat(jdbc.queryForObject("SELECT instance_id FROM
rmq_instance_topic", String.class)).isEmpty();
- assertCode(() -> guard.withInstanceRegistration("cluster", () ->
null), 409);
+ assertThat(jdbc.queryForObject("SELECT domain FROM
rmq_instance_topic", String.class)).isEqualTo("ns:1");
+ assertCode(() -> guard.withInstanceRegistration("first", () -> null),
409);
}
@Test
@@ -556,7 +602,7 @@ class ResourceOwnershipGuardTest {
try (var executor = Executors.newFixedThreadPool(2)) {
java.util.concurrent.Callable<Void> migrate = () -> {
await(start);
- new ResourceOwnershipSchemaMigration(dataSource,
defaultClusters).afterPropertiesSet();
+ new
ResourceOwnershipSchemaMigration(dataSource).afterPropertiesSet();
return null;
};
var a = executor.submit(migrate);
@@ -644,7 +690,7 @@ class ResourceOwnershipGuardTest {
private void legacyIndexes() {
for (String kind : List.of("topic", "group")) {
- jdbc.execute("ALTER TABLE rmq_instance_" + kind + " DROP INDEX
uk_" + kind + "_name");
+ jdbc.execute("ALTER TABLE rmq_instance_" + kind + " DROP INDEX
uk_" + kind + "_domain");
jdbc.execute("CREATE UNIQUE INDEX uk_cluster_instance_" + kind + "
ON rmq_instance_" + kind
+ " (cluster_id, instance_id, name)");
}
@@ -662,7 +708,11 @@ class ResourceOwnershipGuardTest {
private InstanceVO instance(String name) {
InstanceVO instance =
InstanceVO.builder().name(name).vendor(InstanceVendor.APACHE)
- .type(InstanceType.DIRECT).endpoint("same:9876").build();
+ .type(InstanceType.DIRECT)
+ .endpoint(jdbc.queryForObject("SELECT endpoint FROM
rmq_instance WHERE name = ?", String.class, name))
+ .nameserverId(jdbc.queryForObject("SELECT nameserver_id FROM
rmq_instance WHERE name = ?",
+ Long.class, name))
+ .build();
instance.setId(jdbc.queryForObject("SELECT id FROM rmq_instance WHERE
name = ?", Long.class, name));
return instance;
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/LiteTopicServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/LiteTopicServiceTest.java
index d4d2b51c1..daa624b11 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/LiteTopicServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/LiteTopicServiceTest.java
@@ -96,7 +96,7 @@ class LiteTopicServiceTest {
var resource = new
ResourceOwnershipGuard.Resource(ResourceOwnershipGuard.Kind.TOPIC, "chat");
when(guard.requireInstance("instance-a")).thenReturn(instance);
when(guard.topicResource(" chat ")).thenReturn(resource);
- when(guard.check(instance, resource, true)).thenThrow(new
BusinessException(409, "Ownership conflict"));
+ when(guard.checkSlot(instance, resource, null, true)).thenThrow(new
BusinessException(409, "Ownership conflict"));
assertThatThrownBy(() -> service.extendTTL(" instance-a ", " chat ",
60_000L))
.isInstanceOfSatisfying(BusinessException.class, ex ->
assertThat(ex.getCode()).isEqualTo(409));
org.mockito.Mockito.verifyNoInteractions(provider);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
index 6e31a1c8b..ddea630da 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -713,39 +713,21 @@ class MetadataServiceTest {
}
@Test
- void crossInstanceWritesStopBeforeProviderTest() {
- when(ownershipGuard.check(any(), any(),
org.mockito.ArgumentMatchers.anyBoolean()))
+ void crossInstanceDeletesAndMessageOpsStopBeforeProviderTest() {
+ when(ownershipGuard.checkOwnerInDomain(any(), any(),
org.mockito.ArgumentMatchers.anyBoolean()))
.thenThrow(new BusinessException(409, "Resource belongs to
another instance"));
- TopicVO topic = topic("orders", null, TopicType.NORMAL);
- topic.setInstanceId("instance-a");
ConsumerGroupVO group = new ConsumerGroupVO();
group.setName("buyers");
group.setInstanceId("instance-a");
- assertCode(() -> metadataService.createTopic(topic), 409);
- assertCode(() -> metadataService.updateTopic(topic), 409);
+ // Registration writes resolve ownership inside the provider slot
reservation instead,
+ // so only the delete and message paths pre-check here.
assertCode(() -> metadataService.deleteTopic("instance-a", "orders"),
409);
- assertCode(() -> metadataService.createConsumerGroup(group), 409);
- assertCode(() -> metadataService.updateConsumerGroup(group), 409);
assertCode(() -> metadataService.deleteConsumerGroup("instance-a",
"buyers"), 409);
assertCode(() -> metadataService.resetOffset("instance-a", "buyers",
1L, "orders"), 409);
assertCode(() -> metadataService.skipAccumulated("instance-a",
"buyers", "orders"), 409);
assertCode(() -> metadataService.redeliverMessage("instance-a",
"buyers", "orders", "msg", null), 409);
assertCode(() -> metadataService.sendMessage(SendMessageDTO.builder()
.instanceId("instance-a").topic("orders").body("x").build()),
409);
- var topics = metadataService.importTopics("instance-a",
List.of(topicImportRequest("orders", null)));
- assertThat(topics.getFailed()).isEqualTo(1);
- assertThat(topics.getFailures()).singleElement().satisfies(failure -> {
- assertThat(failure.getName()).isEqualTo("orders");
- assertThat(failure.getIndex()).isZero();
- assertThat(failure.getMessage()).isEqualTo("Resource belongs to
another instance");
- });
- var groups = metadataService.importConsumerGroups("instance-a",
List.of(importRequest("buyers", null)));
- assertThat(groups.getFailed()).isEqualTo(1);
- assertThat(groups.getFailures()).singleElement().satisfies(failure -> {
- assertThat(failure.getName()).isEqualTo("buyers");
- assertThat(failure.getIndex()).isZero();
- assertThat(failure.getMessage()).isEqualTo("Resource belongs to
another instance");
- });
verifyNoInteractions(apacheProvider, cloudProvider, adminClient,
runtimeAdminClientResolver, messageService);
}
@@ -1211,7 +1193,7 @@ class MetadataServiceTest {
@Test
void previewResetOffsetShouldRejectForeignGroupBeforeProviderAccess() {
doThrow(new BusinessException(409, "Resource belongs to another
instance"))
- .when(ownershipGuard).check(any(), eq(new
ResourceOwnershipGuard.Resource(
+ .when(ownershipGuard).checkOwnerInDomain(any(), eq(new
ResourceOwnershipGuard.Resource(
ResourceOwnershipGuard.Kind.GROUP, "cg-orders")),
eq(true));
assertThatThrownBy(() -> metadataService.previewResetOffset(
@@ -1224,10 +1206,10 @@ class MetadataServiceTest {
@Test
void previewResetOffsetShouldRejectForeignTopicBeforeProviderAccess() {
- when(ownershipGuard.check(any(), eq(new
ResourceOwnershipGuard.Resource(
+ when(ownershipGuard.checkOwnerInDomain(any(), eq(new
ResourceOwnershipGuard.Resource(
ResourceOwnershipGuard.Kind.GROUP, "cg-orders")),
eq(true))).thenReturn(null);
doThrow(new BusinessException(409, "Resource belongs to another
instance"))
- .when(ownershipGuard).check(any(), eq(new
ResourceOwnershipGuard.Resource(
+ .when(ownershipGuard).checkOwnerInDomain(any(), eq(new
ResourceOwnershipGuard.Resource(
ResourceOwnershipGuard.Kind.TOPIC, "orders")),
eq(true));
assertThatThrownBy(() -> metadataService.previewResetOffset(
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 a59217b0b..6a1fd5947 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
@@ -155,19 +155,35 @@ class RocketMQAdminClientImplTest {
return instance;
});
lenient().when(ownershipGuard.topicResource(any())).thenCallRealMethod();
- lenient().when(ownershipGuard.check(any(), any(),
anyBoolean())).thenAnswer(call -> {
+ lenient().when(ownershipGuard.checkOwnerInDomain(any(), any(),
anyBoolean())).thenAnswer(call -> {
if (!call.<Boolean>getArgument(2)) {
return null;
}
var resource =
call.<org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource>getArgument(1);
return new
org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Ownership(
1L, resource.name(),
call.<org.apache.rocketmq.studio.instance.InstanceVO>getArgument(0).getName(),
- "cluster-1", null);
+ "cluster-1", null, "ns:1");
+ });
+ lenient().when(ownershipGuard.checkSlot(any(), any(), any(),
anyBoolean())).thenAnswer(call -> {
+ if (!call.<Boolean>getArgument(3)) {
+ return null;
+ }
+ var resource =
call.<org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource>getArgument(1);
+ return new
org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Ownership(
+ 1L, resource.name(),
call.<org.apache.rocketmq.studio.instance.InstanceVO>getArgument(0).getName(),
+ "cluster-1", null, "ns:1");
});
lenient().when(ownershipGuard.write(any(), any(), any(), anyBoolean(),
any(), any())).thenAnswer(call ->
call.<java.util.function.Supplier<Object>>getArgument(5).get());
lenient().when(ownershipGuard.withOwned(any(), any(),
any())).thenAnswer(call ->
call.<java.util.function.Supplier<Object>>getArgument(2).get());
+ lenient().when(ownershipGuard.domainOf(any())).thenReturn("ns:1");
+ lenient().when(ownershipGuard.slots(any(), any())).thenAnswer(call -> {
+ var resource =
call.<org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource>getArgument(1);
+ return java.util.List.of(new
org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Ownership(
+ 1L, resource.name(),
call.<org.apache.rocketmq.studio.instance.InstanceVO>getArgument(0).getName(),
+ "cluster-1", null, "ns:1"));
+ });
lenient().when(runtimeAdminClientResolver.execute(anyString(),
any())).thenAnswer(call ->
call.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(adminExt));
lenient().when(runtimeAdminClientResolver.resolveEndpoint(anyString())).thenReturn("10.0.0.1:9876");
@@ -749,7 +765,8 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(topicMapper, times(2)).selectOne(captor.capture());
for (LambdaQueryWrapper<RmqTopic> wrapper : captor.getAllValues()) {
-
assertThat(wrapper.getSqlSegment()).contains("name").doesNotContain("cluster_id",
"instance_id");
+ assertThat(wrapper.getSqlSegment()).contains("name", "domain",
"cluster_id")
+ .doesNotContain("instance_id");
}
}
@@ -858,8 +875,9 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor<LambdaQueryWrapper<RmqTopic>> queries =
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(topicMapper, times(create ? 2 :
1)).selectOne(queries.capture());
for (LambdaQueryWrapper<RmqTopic> query : queries.getAllValues()) {
-
assertThat(query.getSqlSegment()).contains("name").doesNotContain("instance_id",
"cluster_id");
-
assertThat(query.getParamNameValuePairs().values()).containsExactly("orders");
+ assertThat(query.getSqlSegment()).contains("name", "domain",
"cluster_id")
+ .doesNotContain("instance_id");
+
assertThat(query.getParamNameValuePairs().values()).contains("orders", "ns:1");
}
if (create) {
ArgumentCaptor<RmqTopic> saved =
ArgumentCaptor.forClass(RmqTopic.class);
@@ -1232,7 +1250,8 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor<LambdaQueryWrapper<RmqTopic>> captor =
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(topicMapper).delete(captor.capture());
-
assertThat(captor.getValue().getSqlSegment()).contains("name").doesNotContain("cluster_id",
"instance_id");
+ assertThat(captor.getValue().getSqlSegment()).contains("name",
"domain")
+ .doesNotContain("instance_id", "cluster_id");
verify(selectedAdmin).deleteTopicInBroker(Set.of("10.0.0.1:10911"),
"orders");
verify(selectedAdmin).deleteTopicInNameServer(Set.of("10.0.0.2:9876"),
"cluster-1", "orders");
}
@@ -1287,7 +1306,8 @@ class RocketMQAdminClientImplTest {
verify(adminExt,
never()).createAndUpdateSubscriptionGroupConfig(anyString(), any());
ArgumentCaptor<LambdaQueryWrapper<RmqGroup>> captor =
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(groupMapper).selectOne(captor.capture());
-
assertThat(captor.getValue().getSqlSegment()).contains("name").doesNotContain("cluster_id",
"instance_id");
+ assertThat(captor.getValue().getSqlSegment()).contains("name",
"domain", "cluster_id")
+ .doesNotContain("instance_id");
}
@ParameterizedTest
@@ -1662,7 +1682,8 @@ 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("name").doesNotContain("cluster_id",
"instance_id");
+ assertThat(captor.getValue().getSqlSegment()).contains("name",
"domain", "cluster_id")
+ .doesNotContain("instance_id");
}
@Test
@@ -2010,7 +2031,7 @@ class RocketMQAdminClientImplTest {
String destination = targetTopic == null ? "%RETRY%group-a" :
targetTopic;
assertThat(sent.getTopic()).isEqualTo(destination);
var destinationResource = ownershipGuard.topicResource(destination);
- verify(ownershipGuard, org.mockito.Mockito.atLeastOnce()).check(any(),
eq(destinationResource), eq(true));
+ verify(ownershipGuard,
org.mockito.Mockito.atLeastOnce()).checkOwnerInDomain(any(),
eq(destinationResource), eq(true));
MessageExt delivered = new MessageExt();
delivered.setTopic(sent.getTopic());
delivered.setBody(sent.getBody());
@@ -2098,7 +2119,7 @@ class RocketMQAdminClientImplTest {
.body("payload").properties(Map.of(MessageConst.PROPERTY_RETRY_TOPIC,
"orders")).build();
var logicalTopic = new
org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource(
org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind.TOPIC,
"orders");
- when(ownershipGuard.check(any(), eq(logicalTopic), eq(true)))
+ when(ownershipGuard.checkOwnerInDomain(any(), eq(logicalTopic),
eq(true)))
.thenThrow(new
org.apache.rocketmq.studio.common.exception.BusinessException(409, "Topic scope
mismatch"));
assertThatThrownBy(() ->
redeliveryService().redeliverMessage("instance-a", "group-a", source, null))
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
index 3fe083ebb..b3b644ef2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
@@ -93,8 +93,9 @@ class RocketMQLiteTopicProviderTest {
when(guard.requireInstance(any())).thenAnswer(call ->
InstanceVO.builder()
.name(ResourceOwnershipGuard.requireText(call.getArgument(0),
"instanceId")).build());
when(guard.topicResource(anyString())).thenCallRealMethod();
- when(guard.check(any(), any(), eq(true))).thenReturn(new
ResourceOwnershipGuard.Ownership(
- 1L, PARENT, "cluster-a", "cluster-a", "LITE"));
+ when(guard.checkSlot(any(), any(),
org.mockito.ArgumentMatchers.isNull(), eq(true)))
+ .thenReturn(new ResourceOwnershipGuard.Ownership(
+ 1L, PARENT, "cluster-a", "cluster-a", "LITE", "ns:1"));
when(guard.withOwned(any(), any(), any())).thenAnswer(call ->
((java.util.function.Supplier<?>) call.getArgument(2)).get());
doAnswer(call -> ((MqAdminExtFactory.AdminAction<?>)
call.getArgument(1)).apply(admin))
@@ -377,7 +378,7 @@ class RocketMQLiteTopicProviderTest {
.isInstanceOfSatisfying(BusinessException.class, ex ->
assertThat(ex.getCode()).isEqualTo(400));
assertThatThrownBy(() -> provider.extendTTL(" ", PARENT, 60_000L))
.isInstanceOfSatisfying(BusinessException.class, ex ->
assertThat(ex.getCode()).isEqualTo(400));
- when(guard.check(any(), any(), eq(true))).thenThrow(new
BusinessException(409, "Ownership conflict"));
+ when(guard.checkSlot(any(), any(),
org.mockito.ArgumentMatchers.isNull(), eq(true))).thenThrow(new
BusinessException(409, "Ownership conflict"));
assertThatThrownBy(() -> provider.extendTTL("cluster-a", PARENT,
60_000L))
.isInstanceOfSatisfying(BusinessException.class, ex ->
assertThat(ex.getCode()).isEqualTo(409));
org.mockito.Mockito.verifyNoInteractions(runtime, admin);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
index ff4f09fb4..7fc1406c5 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
@@ -213,7 +213,7 @@ class RocketMQMessageProviderTest {
var instance =
org.apache.rocketmq.studio.instance.InstanceVO.builder().name("cluster-a").build();
when(ownershipGuard.requireInstance("instance-a")).thenReturn(instance);
when(ownershipGuard.topicResource("orders")).thenCallRealMethod();
- when(ownershipGuard.check(any(), any(), eq(true)))
+ when(ownershipGuard.checkOwnerInDomain(any(), any(), eq(true)))
.thenThrow(new BusinessException(409, "Resource belongs to
another instance"));
assertDirectConflict();
verify(runtimeAdminClientResolver, never()).execute(anyString(),
any());
@@ -229,9 +229,9 @@ class RocketMQMessageProviderTest {
var instance =
org.apache.rocketmq.studio.instance.InstanceVO.builder().name("cluster-a").build();
when(ownershipGuard.requireInstance("instance-a")).thenReturn(instance);
when(ownershipGuard.topicResource("orders")).thenCallRealMethod();
- when(ownershipGuard.check(any(), any(), eq(true))).thenReturn(
+ when(ownershipGuard.checkOwnerInDomain(any(), any(),
eq(true))).thenReturn(
new
org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Ownership(
- 1L, "orders", "cluster-a", "cluster-a", "NORMAL"));
+ 1L, "orders", "cluster-a", "cluster-a", "NORMAL",
"ns:1"));
when(ownershipGuard.withOwned(any(), any(), any())).thenAnswer(call ->
call.<java.util.function.Supplier<Object>>getArgument(2).get());
ClusterInfo topology = new ClusterInfo();
