You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2026/09/23 03:06:27 UTC
(rocketmq-dashboard) branch rocketmq-studio updated: feat(studio): enforce resource ownership isolation for open source instances (#5006)
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 b42a6ed1b feat(studio): enforce resource ownership isolation for open source instances (#5006)
b42a6ed1b is described below
commit b42a6ed1bdecb1a9fdaecd3e4465c8c1da0c7161
Author: lizhimins <70...@qq.com>
AuthorDate: Wed Sep 23 11:06:21 2026 +0800
feat(studio): enforce resource ownership isolation for open source instances (#5006)
Resolve open-source Apache topic and consumer-group writes to a single
owning instance/cluster and fail closed when the target is ambiguous,
instead of falling back to the first cluster or to all masters/slaves.
- ResourceOwnershipGuard: global name ownership checked under a DB row
lock, with an independent-commit reservation around provider writes
- ApacheWriteTargetResolver: derive the write target from the owner and
reject ambiguous topologies rather than guessing
- ResourceOwnershipSchemaMigration: idempotent startup migration to a
globally unique topic/group name; conflicts fail startup, never drop
- Require a non-blank instanceId on topic/group delete and TTL update
- Guard applies only to Apache instances; cloud vendors stay isolated by
their own instance-scoped APIs
- Cover the guard, resolver, migration and provider paths with tests
---
.../rocketmq/studio/instance/InstanceService.java | 17 +-
.../studio/instance/ResourceOwnershipGuard.java | 287 +++++++++
.../instance/ResourceOwnershipSchemaMigration.java | 153 +++++
.../instance/group/DeleteConsumerGroupDTO.java | 1 +
.../studio/instance/message/MessageService.java | 33 +-
.../studio/instance/topic/DeleteTopicDTO.java | 1 +
.../studio/instance/topic/LiteTopicController.java | 2 +-
.../studio/instance/topic/LiteTopicService.java | 18 +-
.../instance/topic/LiteTopicTTLUpdateDTO.java | 2 +
.../studio/instance/topic/MetadataService.java | 114 ++--
.../rocketmq/studio/provider/InstanceProvider.java | 10 +
.../studio/provider/LiteTopicProvider.java | 6 +
.../studio/provider/apache/AdminClient.java | 2 +
.../provider/apache/ApacheInstanceProvider.java | 23 +
.../provider/apache/ApacheWriteTargetResolver.java | 109 ++++
.../provider/apache/RocketMQAdminClientImpl.java | 401 +++++++-----
.../provider/apache/RocketMQLiteTopicProvider.java | 24 +-
.../provider/apache/RocketMQMessageProvider.java | 70 ++-
.../provider/apache/RocketMQMetadataProvider.java | 97 ++-
server/src/main/resources/db/schema.sql | 35 +-
.../studio/instance/InstanceServiceTest.java | 34 +
.../instance/ResourceOwnershipGuardTest.java | 685 +++++++++++++++++++++
.../group/ConsumerGroupControllerTest.java | 26 +-
.../instance/message/MessageServiceTest.java | 73 ++-
.../instance/topic/LiteTopicControllerTest.java | 25 +-
.../instance/topic/LiteTopicServiceTest.java | 36 +-
.../studio/instance/topic/MetadataServiceTest.java | 201 +++---
.../studio/instance/topic/TopicControllerTest.java | 25 +-
.../ai/tool/service/ToolInstanceRoutingTest.java | 3 +-
.../apache/ApacheInstanceProviderTest.java | 29 +
.../apache/RocketMQAdminClientImplTest.java | 195 ++++--
.../apache/RocketMQLiteTopicProviderTest.java | 55 +-
.../apache/RocketMQMessageProviderTest.java | 93 ++-
.../apache/RocketMQMetadataProviderTest.java | 54 +-
34 files changed, 2378 insertions(+), 561 deletions(-)
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 62a41fdbd..c0c6a3bd7 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
@@ -75,6 +75,7 @@ public class InstanceService {
private final SettingsRepository settingsRepository;
private final CacheManager cacheManager;
private final RegionNames regionNames;
+ private final ResourceOwnershipGuard ownershipGuard;
// @Lazy self-injection: Spring AOP proxies intercept @Transactional calls only when they
// originate from outside the bean. Calling deleteInstance() directly from within this class
@@ -225,7 +226,7 @@ public class InstanceService {
instance.setGmtModified(LocalDateTime.now());
InstanceVO saved;
try {
- saved = instanceRepository.save(instance);
+ saved = ownershipGuard.withInstanceRegistration(instance.getName(), () -> instanceRepository.save(instance));
} catch (DataIntegrityViolationException exception) {
if (vendor != InstanceVendor.APACHE && isCloudCredentialReferenceViolation(exception)) {
throw new BusinessException(409, "Cloud credential no longer exists: " + instance.getCredentialId());
@@ -586,6 +587,7 @@ public class InstanceService {
return StringUtils.hasText(credentialRef) ? credentialRef.trim() : null;
}
+ @Transactional
public InstanceVO updateInstance(InstanceVO instance) {
requireInstance(instance);
log.info("Updating instance: {}", instance.getId());
@@ -630,8 +632,18 @@ public class InstanceService {
}
updated.setGmtModified(LocalDateTime.now());
+ ownershipGuard.lockForInstanceUpdate(existing, updated);
InstanceVO saved = instanceRepository.save(updated);
- releaseApacheClientIfChanged(existing, saved);
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+ @Override
+ public void afterCommit() {
+ releaseApacheClientIfChanged(existing, saved);
+ }
+ });
+ } else {
+ releaseApacheClientIfChanged(existing, saved);
+ }
recordAudit("UPDATE_INSTANCE", "INSTANCE", String.valueOf(saved.getId()), null,
instanceAuditDetail(saved));
return saved;
@@ -645,6 +657,7 @@ public class InstanceService {
throw new BusinessException(400, "InstanceVO ID is required");
}
+ ownershipGuard.lockForInstanceDeletion(id);
InstanceVO existing = instanceRepository.findById(id)
.orElseThrow(() -> new BusinessException(404, "InstanceVO not found: " + id));
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
new file mode 100644
index 000000000..651d8ab2f
--- /dev/null
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.instance;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.springframework.context.annotation.DependsOn;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.dao.PessimisticLockingFailureException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.util.StringUtils;
+
+import java.util.Comparator;
+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. */
+@Component
+@DependsOn("resourceOwnershipSchemaMigration")
+@RequiredArgsConstructor
+public class ResourceOwnershipGuard {
+ private final JdbcTemplate jdbc;
+ private final PlatformTransactionManager transactionManager;
+ private final InstanceResolver instanceResolver;
+
+ @Getter
+ @RequiredArgsConstructor
+ public enum Kind {
+ TOPIC("rmq_instance_topic"), GROUP("rmq_instance_group");
+ private final String table;
+ }
+
+ public record Resource(Kind kind, String name) { }
+
+ public record Ownership(long id, String name, String instanceId, String clusterId, String topicType) { }
+
+ public InstanceVO requireInstance(String identifier) {
+ String id = requireText(identifier, "instanceId");
+ return instanceResolver.findByIdentifier(id)
+ .orElseThrow(() -> new BusinessException(404, "Instance not found: " + id));
+ }
+
+ public static String requireText(String value, String field) {
+ if (!StringUtils.hasText(value)) {
+ throw new BusinessException(400, field + " is required");
+ }
+ return value.trim();
+ }
+
+ /** Derived topics borrow the group ownership only; no regular topic registration is created. */
+ public Resource topicResource(String name) {
+ String topic = requireText(name, "topicName");
+ for (String prefix : List.of("%DLQ%", "%RETRY%")) {
+ if (topic.startsWith(prefix)) {
+ String group = topic.substring(prefix.length());
+ // POP retry topic names have ambiguous separators, so the owning group cannot be guessed.
+ if (!StringUtils.hasText(group) || group.contains("+")) {
+ throw new BusinessException(409, "Cannot determine the owning group of derived topic: " + topic);
+ }
+ return new Resource(Kind.GROUP, group);
+ }
+ }
+ return new Resource(Kind.TOPIC, topic);
+ }
+
+ public Ownership check(InstanceVO instance, Resource resource, boolean required) {
+ return checkOwner(instance, resource, read(resource, false), required);
+ }
+
+ /** Cloud resources are not registered globally yet; reject writes instead of treating OpenAPI visibility as ownership. */
+ public void requireSupportedProvider(InstanceVO instance) {
+ if (instance.getVendor() != null && instance.getVendor() != InstanceVendor.APACHE) {
+ throw new BusinessException(501, "Cloud resources are not registered for global ownership yet; resource writes are not allowed");
+ }
+ }
+
+ /** Direct cloud provider entries also check global conflicts first, then reject unregistered writes. */
+ public void checkProviderWrite(String identifier, String topic, String group) {
+ InstanceVO instance = requireInstance(identifier);
+ if (topic == null && group == null) {
+ throw new BusinessException(400, "resourceName is required");
+ }
+ if (topic != null) {
+ check(instance, topicResource(topic), false);
+ }
+ if (group != null) {
+ check(instance, new Resource(Kind.GROUP, requireText(group, "groupName")), false);
+ }
+ throw new BusinessException(501, "Cloud resources are not registered for global ownership yet; resource writes are not allowed");
+ }
+
+ public <T> T write(InstanceVO instance, Resource resource, String cluster, boolean create,
+ String topicType, Supplier<T> action) {
+ requireSupportedProvider(instance);
+ // 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);
+ 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);
+ } else {
+ jdbc.update("INSERT INTO rmq_instance_group (name, instance_id, cluster_id, status)"
+ + " VALUES (?, ?, ?, 'PENDING')", resource.name(), instance.getName(), cluster);
+ }
+ } 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);
+ 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);
+ requireTopicType(existing, topicType);
+ return action.get();
+ });
+ }
+
+ /** Multi-resource operations lock in a fixed order so the guard can be reused within one transaction. */
+ public <T> T withOwned(InstanceVO instance, List<Resource> resources, Supplier<T> action) {
+ requireSupportedProvider(instance);
+ return new TransactionTemplate(transactionManager).execute(status -> {
+ 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));
+ return action.get();
+ });
+ }
+
+ /** Must lock the instance inside the deletion transaction first, then re-read claims in all states. */
+ public void lockForInstanceDeletion(Long id) {
+ if (!TransactionSynchronizationManager.isActualTransactionActive()) {
+ throw new IllegalStateException("Instance deletion must run inside a database transaction");
+ }
+ List<String> names = jdbc.query("SELECT name FROM rmq_instance WHERE id = ?",
+ (rs, row) -> rs.getString(1), id);
+ if (names.isEmpty()) {
+ throw new BusinessException(404, "Instance not found: " + id);
+ }
+ lockIdentity(names.getFirst());
+ if (jdbc.query("SELECT id FROM rmq_instance WHERE id = ? FOR UPDATE",
+ (rs, row) -> rs.getLong(1), id).isEmpty()) {
+ throw new BusinessException(404, "Instance not found: " + id);
+ }
+ requireUnoccupiedInstance(names.getFirst());
+ }
+
+ /** Connection changes are serialized with resource writes; occupied instances cannot point to another endpoint. */
+ public void lockForInstanceUpdate(InstanceVO existing, InstanceVO updated) {
+ if (!TransactionSynchronizationManager.isActualTransactionActive()) {
+ throw new IllegalStateException("Instance update must run inside a database transaction");
+ }
+ lockInstance(existing);
+ if (!Objects.equals(existing.getEndpoint(), updated.getEndpoint())
+ || !Objects.equals(existing.getType(), updated.getType())) {
+ lockForInstanceDeletion(existing.getId());
+ }
+ }
+
+ /** Registration and virtual instance claiming share one mutex row; gap locks on absent rows are not reliable. */
+ public <T> T withInstanceRegistration(String name, Supplier<T> registration) {
+ String identifier = requireText(name, "instanceId");
+ return new TransactionTemplate(transactionManager).execute(status -> {
+ lockIdentity(identifier);
+ requireUnoccupiedInstance(identifier);
+ return registration.get();
+ });
+ }
+
+ private void requireUnoccupiedInstance(String name) {
+ for (Kind kind : Kind.values()) {
+ if (!jdbc.query("SELECT id FROM " + kind.table
+ + " WHERE instance_id = ? OR ((instance_id IS NULL OR TRIM(instance_id) = '')"
+ + " AND cluster_id = ?) FOR UPDATE", (rs, row) -> rs.getLong(1), name, name).isEmpty()) {
+ throw new BusinessException(409, "Instance name still holds resources or unfinished claims; cannot register, delete or change connection: " + name);
+ }
+ }
+ }
+
+ private void lockIdentity(String name) {
+ jdbc.update("INSERT INTO rmq_instance_ownership_lock (name) VALUES (?) ON DUPLICATE KEY UPDATE name = name", name);
+ jdbc.queryForObject("SELECT id FROM rmq_instance_ownership_lock WHERE name = ? FOR UPDATE", Long.class, name);
+ }
+
+ private void lockInstance(InstanceVO instance) {
+ 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()) {
+ throw new BusinessException(409, "Virtual instance is shadowed by a registered instance; please re-confirm ownership");
+ }
+ return;
+ }
+ List<Boolean> matches = jdbc.query("SELECT name, endpoint, admin_credential_ref, type, vendor FROM rmq_instance"
+ + " WHERE id = ? FOR UPDATE", (rs, row) -> instance.getName().equals(rs.getString(1))
+ && Objects.equals(instance.getEndpoint(), rs.getString(2))
+ && Objects.equals(instance.getAdminCredentialRef(), rs.getString(3))
+ && Objects.equals(instance.getType() == null ? null : instance.getType().name(), rs.getString(4))
+ && Objects.equals(instance.getVendor() == null ? InstanceVendor.APACHE.name() : instance.getVendor().name(),
+ StringUtils.hasText(rs.getString(5)) ? rs.getString(5) : InstanceVendor.APACHE.name()),
+ instance.getId());
+ if (matches.isEmpty() || !matches.getFirst()) {
+ throw new BusinessException(409, "Instance was deleted or changed; please re-confirm ownership");
+ }
+ }
+
+ private Ownership read(Resource resource, 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());
+ }
+ return rows.isEmpty() ? null : rows.getFirst();
+ }
+
+ private Ownership checkOwner(InstanceVO instance, Resource resource, Ownership owner, boolean required) {
+ if (owner == null) {
+ if (required) {
+ throw new BusinessException(404, "Resource ownership is not registered yet: " + resource.name());
+ }
+ return null;
+ }
+ boolean legacyVirtual = !StringUtils.hasText(owner.instanceId()) && instance.getId() == null
+ && instance.getName().equals(owner.clusterId());
+ if (!instance.getName().equals(owner.instanceId()) && !legacyVirtual) {
+ 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");
+ }
+ }
+
+ private TransactionTemplate newTransaction() {
+ TransactionTemplate transaction = new TransactionTemplate(transactionManager);
+ transaction.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+ return transaction;
+ }
+}
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
new file mode 100644
index 000000000..6cfbfab7c
--- /dev/null
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.instance;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.rocketmq.studio.provider.apache.RocketMQDefaultClusterResolver;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.boot.sql.init.dependency.DependsOnDatabaseInitialization;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Validates the migration after SQL initialization and before write services start; never deduplicates or rewrites ownership. */
+@Component
+@DependsOnDatabaseInitialization
+@RequiredArgsConstructor
+public class ResourceOwnershipSchemaMigration implements InitializingBean {
+ 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.
+ 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));
+ }
+ }
+ }
+ 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));
+ }
+ }
+ }
+ }
+ }
+ statement.executeUpdate("CREATE TABLE IF NOT EXISTS rmq_instance_ownership_lock ("
+ + "id bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'primary key',"
+ + "gmt_create datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'create time',"
+ + "gmt_modified datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'modify time',"
+ + "name VARCHAR(128) NOT NULL, PRIMARY KEY (id), UNIQUE KEY uk_ownership_instance_name (name))"
+ + " 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;
+ }
+ }
+ }
+ }
+ }
+ }
+
+ private void ensureIndex(Connection connection, String table, String name, String column, boolean unique)
+ throws SQLException {
+ Map<String, List<String>> indexes = indexes(connection, table);
+ String expected = (unique ? "unique:" : "index:") + column;
+ if (indexes.containsKey(name)) {
+ if (!indexes.get(name).equals(List.of(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 + ")");
+ } 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))) {
+ throw failure;
+ }
+ }
+ }
+
+ private Map<String, List<String>> indexes(Connection connection, String table) throws SQLException {
+ Map<String, List<String>> indexes = new HashMap<>();
+ try (ResultSet rows = connection.getMetaData().getIndexInfo(connection.getCatalog(), null, table, false, false)) {
+ while (rows.next()) {
+ 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<>())
+ .add((rows.getBoolean("NON_UNIQUE") ? "index:" : "unique:")
+ + column.toLowerCase(java.util.Locale.ROOT));
+ }
+ }
+ }
+ return indexes;
+ }
+}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/instance/group/DeleteConsumerGroupDTO.java b/server/src/main/java/org/apache/rocketmq/studio/instance/group/DeleteConsumerGroupDTO.java
index 5a3e5caa5..fd3116d8f 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/group/DeleteConsumerGroupDTO.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/group/DeleteConsumerGroupDTO.java
@@ -30,5 +30,6 @@ public class DeleteConsumerGroupDTO {
@NotBlank(message = "name is required")
private String name;
+ @NotBlank(message = "instanceId is required")
private String instanceId;
}
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 ce62474e3..cb85a5528 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
@@ -18,7 +18,11 @@ package org.apache.rocketmq.studio.instance.message;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.springframework.stereotype.Service;
@@ -39,6 +43,7 @@ public class MessageService {
private final InstanceProviderRegistry providerRegistry;
private final QueryHistoryService queryHistoryService;
private final OperationAuditService operationAuditService;
+ private final ResourceOwnershipGuard ownershipGuard;
public List<MessageRecordVO> queryMessages(
String instanceId, String topic, String msgId, String tag, String key, Long startTime, Long endTime) {
@@ -134,12 +139,34 @@ public class MessageService {
}
public DirectConsumeMessageResultVO consumeMessageDirectly(DirectConsumeMessageDTO request) {
+ if (request == null) {
+ throw new BusinessException(400, "Direct consume request must not be null");
+ }
+ var instance = ownershipGuard.requireInstance(request.getInstanceId());
+ request.setInstanceId(instance.getName());
+ request.setTopic(ResourceOwnershipGuard.requireText(request.getTopic(), "topicName"));
+ request.setConsumerGroup(ResourceOwnershipGuard.requireText(request.getConsumerGroup(), "groupName"));
+ request.setClientId(ResourceOwnershipGuard.requireText(request.getClientId(), "clientId"));
+ request.setMsgId(ResourceOwnershipGuard.requireText(request.getMsgId(), "msgId"));
+ // Ownership guard applies to open-source Apache instances only; cloud calls are isolated by their own instance id.
+ boolean apache = instance.getVendor() == null || instance.getVendor() == InstanceVendor.APACHE;
+ 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));
+ ownershipGuard.requireSupportedProvider(instance);
+ }
String detail = "topic=" + request.getTopic() + ", consumerGroup=" + request.getConsumerGroup()
+ ", clientId=" + request.getClientId();
try {
- DirectConsumeMessageResultVO result = providerRegistry.byInstanceId(request.getInstanceId())
- .map(provider -> provider.consumeMessageDirectly(request))
- .orElseGet(() -> messageProvider.consumeMessageDirectly(request));
+ DirectConsumeMessageResultVO result = apache
+ ? ownershipGuard.withOwned(instance, resources, () ->
+ providerRegistry.byInstanceId(request.getInstanceId())
+ .map(provider -> provider.consumeMessageDirectly(request))
+ .orElseGet(() -> messageProvider.consumeMessageDirectly(request)))
+ : providerRegistry.byInstanceId(request.getInstanceId())
+ .map(provider -> provider.consumeMessageDirectly(request))
+ .orElseGet(() -> messageProvider.consumeMessageDirectly(request));
recordDirectConsumeAudit(request, detail + ", result=" + result.getConsumeResult(),
auditResult(result.getConsumeResult()), null);
return result;
diff --git a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
index 746e887c7..3f4a34906 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
@@ -30,5 +30,6 @@ public class DeleteTopicDTO {
@NotBlank(message = "name is required")
private String name;
+ @NotBlank(message = "instanceId is required")
private String instanceId;
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicController.java b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicController.java
index ff98e498c..7526a2505 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicController.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicController.java
@@ -51,7 +51,7 @@ public class LiteTopicController {
@PostMapping("/extendTTL")
public Result<Void> extendTTL(@Valid @RequestBody LiteTopicTTLUpdateDTO request) {
- liteTopicService.extendTTL(request.getTopicPattern(), request.getNewTTL());
+ liteTopicService.extendTTL(request.getInstanceId(), request.getTopicPattern(), request.getNewTTL());
return Result.ok();
}
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 caf9cf7f0..97ddf7897 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
@@ -17,7 +17,9 @@
package org.apache.rocketmq.studio.instance.topic;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
import org.apache.rocketmq.studio.model.LiteTopicQuota;
import org.apache.rocketmq.studio.model.LiteTopicSession;
import org.apache.rocketmq.studio.model.LiteTopicSummary;
@@ -42,6 +44,7 @@ import java.util.List;
public class LiteTopicService {
private final LiteTopicProvider liteTopicProvider;
+ private final ResourceOwnershipGuard ownershipGuard;
public List<LiteTopicItemVO> listLiteTopics(String pattern, String namespace) {
return liteTopicProvider.listLiteTopics(pattern, namespace).stream()
@@ -56,14 +59,25 @@ public class LiteTopicService {
return toSessionVO(liteTopicProvider.getSession(sessionId));
}
- public void extendTTL(String topicPattern, Long newTTL) {
+ public void extendTTL(String instanceId, String topicPattern, Long newTTL) {
+ String identifier = ResourceOwnershipGuard.requireText(instanceId, "instanceId");
if (topicPattern == null || topicPattern.isBlank()) {
throw new BusinessException(400, "topicPattern is required");
}
if (newTTL == null || newTTL <= 0) {
throw new BusinessException(400, "newTTL must be positive");
}
- liteTopicProvider.extendTTL(topicPattern.trim(), newTTL);
+ var instance = ownershipGuard.requireInstance(identifier);
+ 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.requireSupportedProvider(instance);
+ }
+ if (resource.kind() != ResourceOwnershipGuard.Kind.TOPIC) {
+ throw new BusinessException(409, "TTL can only be updated on a registered Lite parent topic");
+ }
+ liteTopicProvider.extendTTL(instance.getName(), resource.name(), newTTL);
}
public LiteTopicQuotaVO getQuota(String namespace) {
diff --git a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicTTLUpdateDTO.java b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicTTLUpdateDTO.java
index 2e37c68e3..af79f07e6 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicTTLUpdateDTO.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/LiteTopicTTLUpdateDTO.java
@@ -24,6 +24,8 @@ import lombok.Data;
@Data
public class LiteTopicTTLUpdateDTO {
+ @NotBlank(message = "instanceId is required")
+ private String instanceId;
@NotBlank(message = "topicPattern is required")
private String topicPattern;
@NotNull(message = "newTTL is required")
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 06a9a11bb..dbd5dace1 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
@@ -29,6 +29,10 @@ import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.util.MqResponseCodes;
import org.apache.rocketmq.studio.instance.InstanceResolver;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
import org.apache.rocketmq.studio.provider.apache.AdminClient;
import org.apache.rocketmq.studio.provider.apache.ConsumerLagResolver;
import org.apache.rocketmq.studio.provider.apache.MetadataProvider;
@@ -82,6 +86,7 @@ public class MetadataService {
private final OperationAuditService operationAuditService;
private final MessageService messageService;
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
+ private final ResourceOwnershipGuard ownershipGuard;
/**
* Canonicalizes registered instance names and legacy numeric IDs, while preserving physical
@@ -151,14 +156,20 @@ public class MetadataService {
}
public TopicVO createTopic(String instanceId, TopicVO topic) {
+ return saveTopic(instanceId, topic, false);
+ }
+
+ private TopicVO saveTopic(String instanceId, TopicVO topic, boolean importing) {
requireTopic(topic);
if (SystemTopicFilter.isSystem(topic.getName())) {
throw new BusinessException(400, "System topics cannot be created: " + topic.getName());
}
- topic.setInstanceId(instanceId);
- InstanceProvider provider = resolve(instanceId);
+ String target = requireWriteInstance(instanceId, new Resource(Kind.TOPIC, topic.getName()), false);
+ topic.setInstanceId(target);
+ InstanceProvider provider = resolve(target);
return executeWithAudit(provider, Operation.CREATE_TOPIC, ResourceType.TOPIC, topic.getName(),
- instanceId, topicDetail(topic), () -> provider.createTopic(instanceId, topic));
+ target, topicDetail(topic), () -> importing
+ ? provider.importTopic(target, topic) : provider.createTopic(target, topic));
}
public TopicVO updateTopic(TopicVO topic) {
@@ -167,11 +178,12 @@ public class MetadataService {
public TopicVO updateTopic(String instanceId, TopicVO topic) {
requireTopic(topic);
- topic.setInstanceId(instanceId);
- guardImmutableType(instanceId, topic);
- InstanceProvider provider = resolve(instanceId);
+ String target = requireWriteInstance(instanceId, new Resource(Kind.TOPIC, topic.getName()), false);
+ topic.setInstanceId(target);
+ guardImmutableType(target, topic);
+ InstanceProvider provider = resolve(target);
return executeWithAudit(provider, Operation.UPDATE_TOPIC, ResourceType.TOPIC, topic.getName(),
- instanceId, topicDetail(topic), () -> provider.updateTopic(instanceId, topic));
+ target, topicDetail(topic), () -> provider.updateTopic(target, topic));
}
/** The registered message type of an existing topic is immutable (creation-only attribute). */
@@ -191,8 +203,8 @@ public class MetadataService {
}
public void deleteTopic(String instanceId, String name) {
- String target = normalizeInstanceId(instanceId);
String topicName = requireName(name, "topic name");
+ String target = requireWriteInstance(instanceId, ownershipGuard.topicResource(topicName), true);
InstanceProvider provider = resolve(target);
executeWithAudit(provider, Operation.DELETE_TOPIC, ResourceType.TOPIC,
topicName, target, null, () -> provider.deleteTopic(target, topicName));
@@ -280,6 +292,9 @@ public class MetadataService {
public SendMessageVO sendMessage(SendMessageDTO request) {
requireSendMessageRequest(request);
+ request.setTopic(requireName(request.getTopic(), "topicName"));
+ request.setInstanceId(requireWriteInstance(request.getInstanceId(),
+ ownershipGuard.topicResource(request.getTopic()), true));
return resolve(request.getInstanceId()).sendMessage(request);
}
@@ -293,25 +308,37 @@ public class MetadataService {
public SendMessageVO redeliverMessage(String instanceId, String groupName, String sourceTopic,
String msgId, String targetTopic) {
String group = requireName(groupName, "group name");
- MessageRecordVO original = findMessageForRedelivery(instanceId, sourceTopic, msgId);
- return redeliverMessage(instanceId, group, original, targetTopic);
+ String target = requireWriteInstance(instanceId, new Resource(Kind.GROUP, group), true);
+ if (StringUtils.hasText(sourceTopic)) {
+ requireWriteInstance(target, ownershipGuard.topicResource(sourceTopic), true);
+ }
+ MessageRecordVO original = findMessageForRedelivery(target, sourceTopic, msgId);
+ return redeliverMessage(target, group, original, targetTopic);
}
public SendMessageVO redeliverMessage(String instanceId, String groupName, MessageRecordVO original,
String targetTopic) {
String group = requireName(groupName, "group name");
+ String target = requireWriteInstance(instanceId, new Resource(Kind.GROUP, group), true);
+ if (original == null) {
+ throw new BusinessException(400, "Message to redeliver must not be null");
+ }
String destination = StringUtils.hasText(targetTopic)
? targetTopic.trim()
: MixAll.getRetryTopic(group);
SendMessageDTO request = SendMessageDTO.builder()
- .instanceId(normalizeInstanceId(instanceId))
+ .instanceId(target)
.topic(destination)
.tag(original.getTag())
.key(original.getKey())
.body(original.getBody())
.properties(redeliveryProperties(original.getProperties()))
.build();
- return sendMessage(request);
+ InstanceVO instance = ownershipGuard.requireInstance(target);
+ List<Resource> resources = List.of(new Resource(Kind.GROUP, group),
+ ownershipGuard.topicResource(original.getTopic()), ownershipGuard.topicResource(destination));
+ resources.forEach(resource -> ownershipGuard.check(instance, resource, true));
+ return ownershipGuard.withOwned(instance, resources, () -> sendMessage(request));
}
/** Drops the system-reserved keys {@code Message.putUserProperty} would reject (ยง15.5.1 KEYS defect). */
@@ -474,13 +501,17 @@ public class MetadataService {
private ConsumerGroupVO saveConsumerGroup(String instanceId, ConsumerGroupVO group, String operation,
BiFunction<InstanceProvider, String, ConsumerGroupVO> mutation) {
- if (group != null) {
- group.setInstanceId(instanceId);
+ ResourceOwnershipGuard.requireText(instanceId, "instanceId");
+ if (group == null) {
+ throw new BusinessException(400, "Group request is required");
}
- InstanceProvider provider = resolve(instanceId);
+ group.setName(requireName(group.getName(), "groupName"));
+ String target = requireWriteInstance(instanceId, new Resource(Kind.GROUP, group.getName()),
+ Operation.UPDATE_GROUP.equals(operation));
+ group.setInstanceId(target);
+ InstanceProvider provider = resolve(target);
return executeWithAudit(provider, operation, ResourceType.GROUP,
- group == null ? null : group.getName(), instanceId, consumerGroupDetail(group),
- () -> mutation.apply(provider, instanceId));
+ group.getName(), target, consumerGroupDetail(group), () -> mutation.apply(provider, target));
}
public ConsumerGroupSettingsVO getConsumerGroupSettings(String instanceId, String name) {
@@ -491,10 +522,9 @@ public class MetadataService {
public ConsumerGroupSettingsVO updateConsumerGroupSettings(String instanceId, String name,
ConsumerGroupSettingsCommand command) {
- instanceId = normalizeInstanceId(instanceId);
- requireApacheInstance(instanceId);
String groupName = requireName(name, "consumer group name");
- return adminClient.updateConsumerGroupSettings(instanceId, groupName, command);
+ String target = requireWriteInstance(instanceId, new Resource(Kind.GROUP, groupName), true);
+ return adminClient.updateConsumerGroupSettings(target, groupName, command);
}
@@ -503,7 +533,8 @@ public class MetadataService {
}
public void deleteConsumerGroup(String instanceId, String name) {
- String target = normalizeInstanceId(instanceId);
+ String target = requireWriteInstance(instanceId,
+ new Resource(Kind.GROUP, requireName(name, "groupName")), true);
deleteConsumerGroup(target, name, (provider, groupName) -> provider.deleteConsumerGroup(target, groupName));
}
@@ -512,19 +543,7 @@ public class MetadataService {
InstanceProvider provider = resolve(instanceId);
executeWithAudit(provider, Operation.DELETE_GROUP, ResourceType.GROUP,
groupName, instanceId, null, () -> mutation.accept(provider, groupName));
- cascadeDeleteDlqTopic(instanceId, groupName);
- }
-
- /** Best-effort DLQ cascade (decision 17): a missing or undeletable %DLQ% topic never blocks group deletion. */
- private void cascadeDeleteDlqTopic(String instanceId, String groupName) {
- String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
- try {
- deleteTopic(instanceId, dlqTopic);
- log.info("Cascaded DLQ topic deletion for consumer group {}: {}", groupName, dlqTopic);
- } catch (Exception e) {
- log.warn("Failed to cascade delete DLQ topic {} for consumer group {}: {}",
- dlqTopic, groupName, e.getMessage());
- }
+ // Derived topic cleanup is handled by the Apache layer inside the group ownership lock.
}
public void resetOffset(String name, long timestamp, String topic) {
@@ -540,9 +559,10 @@ public class MetadataService {
}
public void resetOffset(String instanceId, String name, long timestamp, String topic) {
- instanceId = normalizeInstanceId(instanceId);
String groupName = requireName(name, "consumer group name");
String topicName = requireName(topic, "topic name");
+ instanceId = requireWriteInstance(instanceId, new Resource(Kind.GROUP, groupName), true);
+ requireWriteInstance(instanceId, ownershipGuard.topicResource(topicName), true);
InstanceProvider provider = resolve(instanceId);
String normalizedInstanceId = instanceId;
executeWithAudit(provider, Operation.RESET_OFFSET, ResourceType.GROUP, groupName, instanceId,
@@ -556,9 +576,12 @@ public class MetadataService {
*/
public List<String> skipAccumulated(String instanceId, String name, String topic) {
String groupName = requireName(name, "consumer group name");
- List<String> topics = resolveSkipAccumulatedTopics(instanceId, groupName, topic);
+ String target = requireWriteInstance(instanceId, new Resource(Kind.GROUP, groupName), true);
+ List<String> topics = resolveSkipAccumulatedTopics(target, groupName, topic);
+ topics.forEach(subscriptionTopic -> requireWriteInstance(target,
+ ownershipGuard.topicResource(subscriptionTopic), true));
long timestamp = System.currentTimeMillis();
- topics.forEach(subscriptionTopic -> resetOffset(instanceId, groupName, timestamp, subscriptionTopic));
+ topics.forEach(subscriptionTopic -> resetOffset(target, groupName, timestamp, subscriptionTopic));
return topics;
}
@@ -612,6 +635,18 @@ public class MetadataService {
if (topic == null) {
throw new BusinessException(400, "Topic request is required");
}
+ topic.setName(requireName(topic.getName(), "topicName"));
+ }
+
+ private String requireWriteInstance(String instanceId, Resource resource, boolean required) {
+ ResourceOwnershipGuard.requireText(instanceId, "instanceId");
+ 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);
+ ownershipGuard.requireSupportedProvider(instance);
+ }
+ return instance.getName();
}
private void requireSendMessageRequest(SendMessageDTO request) {
@@ -646,7 +681,8 @@ public class MetadataService {
}
ConsumerGroupVO group = request.toConsumerGroupVO();
group.setInstanceId(normalizedInstanceId);
- imported.add(createConsumerGroup(group));
+ imported.add(saveConsumerGroup(normalizedInstanceId, group, Operation.CREATE_GROUP,
+ (provider, target) -> provider.importConsumerGroup(target, group)));
} catch (Exception exception) {
failures.add(ImportConsumerGroupsResultVO.Failure.builder()
.index(index)
@@ -747,7 +783,7 @@ public class MetadataService {
}
TopicVO topic = request.toTopicVO();
topic.setInstanceId(normalizedInstanceId);
- imported.add(createTopic(topic));
+ imported.add(saveTopic(normalizedInstanceId, topic, true));
} catch (Exception exception) {
failures.add(ImportTopicsResultVO.Failure.builder()
.index(index)
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
index d419ca7de..eb596c75b 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
@@ -80,6 +80,11 @@ public interface InstanceProvider {
TopicVO createTopic(String instanceId, TopicVO topic);
+ /** Imports an existing topic; defaults to creation, providers may override to compare configurations. */
+ default TopicVO importTopic(String instanceId, TopicVO topic) {
+ return createTopic(instanceId, topic);
+ }
+
TopicVO updateTopic(String instanceId, TopicVO topic);
void deleteTopic(String instanceId, String topicName);
@@ -123,6 +128,11 @@ public interface InstanceProvider {
ConsumerGroupVO createConsumerGroup(String instanceId, ConsumerGroupVO group);
+ /** Imports an existing consumer group; defaults to creation, providers may override to compare configurations. */
+ default ConsumerGroupVO importConsumerGroup(String instanceId, ConsumerGroupVO group) {
+ return createConsumerGroup(instanceId, group);
+ }
+
default ConsumerGroupVO updateConsumerGroup(String instanceId, ConsumerGroupVO group) {
return createConsumerGroup(instanceId, group);
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/LiteTopicProvider.java b/server/src/main/java/org/apache/rocketmq/studio/provider/LiteTopicProvider.java
index 7281cfd19..87a976db4 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/LiteTopicProvider.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/LiteTopicProvider.java
@@ -76,6 +76,12 @@ public interface LiteTopicProvider {
* @param ttlMillis the new TTL in milliseconds
*/
default void extendTTL(String topicPattern, long ttlMillis) {
+ throw new BusinessException(400, "instanceId is required");
+ }
+
+ /** TTL writes must explicitly specify the owning instance; the legacy entry point must not fall back to the default connection. */
+ default void extendTTL(String instanceId, String topicPattern, long ttlMillis) {
+ org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.requireText(instanceId, "instanceId");
throw new BusinessException(NOT_IMPLEMENTED, UNSUPPORTED);
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
index 275ea7e5b..49f96e84e 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
@@ -37,10 +37,12 @@ public interface AdminClient {
}
TopicVO createTopic(String instanceId, TopicVO topic);
+ TopicVO importTopic(String instanceId, TopicVO topic);
TopicVO updateTopic(String instanceId, TopicVO topic);
void deleteTopic(String instanceId, String name);
SendMessageVO sendMessage(SendMessageDTO request);
ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group);
+ ConsumerGroupVO importConsumerGroup(ConsumerGroupVO group);
ConsumerGroupVO updateConsumerGroup(ConsumerGroupVO group);
ConsumerGroupSettingsVO getConsumerGroupSettings(String instanceId, String name);
ConsumerGroupSettingsVO updateConsumerGroupSettings(String instanceId, String name,
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
index faea84e07..f2c5572d6 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
@@ -19,6 +19,8 @@ package org.apache.rocketmq.studio.provider.apache;
import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
@@ -108,6 +110,11 @@ public class ApacheInstanceProvider implements InstanceProvider {
return adminClient.createTopic(instanceId, topic);
}
+ @Override
+ public TopicVO importTopic(String instanceId, TopicVO topic) {
+ return adminClient.importTopic(instanceId, topic);
+ }
+
@Override
public TopicVO updateTopic(String instanceId, TopicVO topic) {
return adminClient.updateTopic(instanceId, topic);
@@ -147,14 +154,30 @@ public class ApacheInstanceProvider implements InstanceProvider {
@Override
public ConsumerGroupVO createConsumerGroup(String instanceId, ConsumerGroupVO group) {
+ requireGroupInstance(instanceId, group);
return adminClient.createConsumerGroup(group);
}
+ @Override
+ public ConsumerGroupVO importConsumerGroup(String instanceId, ConsumerGroupVO group) {
+ requireGroupInstance(instanceId, group);
+ return adminClient.importConsumerGroup(group);
+ }
+
@Override
public ConsumerGroupVO updateConsumerGroup(String instanceId, ConsumerGroupVO group) {
+ requireGroupInstance(instanceId, group);
return adminClient.updateConsumerGroup(group);
}
+ private void requireGroupInstance(String instanceId, ConsumerGroupVO group) {
+ String target = ResourceOwnershipGuard.requireText(instanceId, "instanceId");
+ if (group == null) {
+ throw new BusinessException(400, "Group request is required");
+ }
+ group.setInstanceId(target);
+ }
+
@Override
public void deleteConsumerGroup(String instanceId, String groupName) {
adminClient.deleteConsumerGroup(instanceId, groupName);
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheWriteTargetResolver.java b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheWriteTargetResolver.java
new file mode 100644
index 000000000..d20f14a5f
--- /dev/null
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheWriteTargetResolver.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider.apache;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.springframework.util.StringUtils;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** A write target must come from trusted ownership or a unique topology; never guess the physical cluster from the instance name. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class ApacheWriteTargetResolver {
+ public record Target(String cluster, Set<String> masters, Set<String> brokers) { }
+
+ public static Target resolve(MQAdminExt admin, InstanceVO instance, String ownedCluster) throws Exception {
+ ClusterInfo info = admin.examineBrokerClusterInfo();
+ if (info == null || info.getClusterAddrTable() == null || info.getBrokerAddrTable() == null
+ || info.getClusterAddrTable().isEmpty()) {
+ throw new BusinessException(409, "Incomplete cluster topology; write rejected");
+ }
+ String cluster = ownedCluster;
+ if (instance.getId() == null) {
+ if (StringUtils.hasText(cluster) && !instance.getName().equals(cluster)) {
+ throw new BusinessException(409, "Virtual instance does not match the physical cluster of the resource");
+ }
+ cluster = instance.getName();
+ } else if (!StringUtils.hasText(cluster)) {
+ if (info.getClusterAddrTable().size() != 1) {
+ throw new BusinessException(409, "Registered instance is not bound to an explicit physical cluster; writes are rejected on multi-cluster endpoints");
+ }
+ cluster = info.getClusterAddrTable().keySet().iterator().next();
+ for (BrokerData broker : info.getBrokerAddrTable().values()) {
+ if (!StringUtils.hasText(cluster) || broker == null || !cluster.equals(broker.getCluster())) {
+ throw new BusinessException(409, "Cluster table and broker table are inconsistent; topology is not uniquely determined");
+ }
+ }
+ }
+ return target(info, cluster);
+ }
+
+ public static Target target(ClusterInfo info, String cluster) {
+ if (!StringUtils.hasText(cluster) || info == null || info.getClusterAddrTable() == null
+ || info.getBrokerAddrTable() == null) {
+ throw new BusinessException(409, "Cannot determine the target cluster; write rejected");
+ }
+ Set<String> names = info.getClusterAddrTable().get(cluster);
+ if (names == null || names.isEmpty()) {
+ throw new BusinessException(409, "Target cluster does not exist or has no broker: " + cluster);
+ }
+ for (var entry : info.getBrokerAddrTable().entrySet()) {
+ BrokerData broker = entry.getValue();
+ if (broker != null && cluster.equals(broker.getCluster()) && !names.contains(entry.getKey())) {
+ throw new BusinessException(409, "Target cluster is missing a broker; incomplete topology rejected: " + entry.getKey());
+ }
+ }
+ Set<String> masters = new HashSet<>();
+ for (String name : names) {
+ BrokerData broker = info.getBrokerAddrTable().get(name);
+ if (!StringUtils.hasText(name) || broker == null || !cluster.equals(broker.getCluster())
+ || !name.equals(broker.getBrokerName())
+ || broker.getBrokerAddrs() == null || !StringUtils.hasText(broker.getBrokerAddrs().get(0L))) {
+ throw new BusinessException(409, "Incomplete broker/master topology in the target cluster: " + name);
+ }
+ masters.add(broker.getBrokerAddrs().get(0L));
+ }
+ return new Target(cluster, Set.copyOf(masters), Set.copyOf(names));
+ }
+
+ /** Name-addressed send/reset RPCs must reject cross-cluster aggregated routes so the SDK cannot fan out on its own. */
+ public static void requireTopicRoute(MQAdminExt admin, Target target, String topic) throws Exception {
+ var route = admin.examineTopicRouteInfo(topic);
+ if (route == null || route.getBrokerDatas() == null || route.getBrokerDatas().isEmpty()) {
+ throw new BusinessException(409, "Incomplete topic route; resource write rejected: " + topic);
+ }
+ for (BrokerData broker : route.getBrokerDatas()) {
+ if (broker == null || !target.cluster().equals(broker.getCluster())
+ || !target.brokers().contains(broker.getBrokerName()) || broker.getBrokerAddrs() == null
+ || !target.masters().contains(broker.getBrokerAddrs().get(0L))) {
+ throw new BusinessException(409, "Topic route contains a broker outside the target cluster: " + topic);
+ }
+ }
+ if (route.getQueueDatas() == null || route.getQueueDatas().isEmpty()
+ || route.getQueueDatas().stream().anyMatch(queue -> queue == null
+ || !target.brokers().contains(queue.getBrokerName()))) {
+ throw new BusinessException(409, "Incomplete topic queue route: " + topic);
+ }
+ }
+}
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 16553a9ac..72f2b3671 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
@@ -39,6 +39,10 @@ import org.apache.rocketmq.studio.common.util.MqResponseCodes;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupSettingsCommand;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupSettingsVO;
@@ -82,7 +86,6 @@ import java.util.Set;
public class RocketMQAdminClientImpl implements AdminClient {
private static final int MAX_MESSAGE_SIZE = 4 * 1024 * 1024; // 4 MB default broker limit
- private static final String LEGACY_METADATA_SCOPE = "";
private static final long RESET_OFFSET_PREVIEW_TIMEOUT_MILLIS = 3_000L;
private static final String RISK_INFO = "INFO";
private static final String RISK_WARNING = "WARNING";
@@ -95,6 +98,7 @@ public class RocketMQAdminClientImpl implements AdminClient {
private final AuditService auditService;
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
private final MqClientPool clientPool;
+ private final ResourceOwnershipGuard ownershipGuard;
@org.springframework.beans.factory.annotation.Autowired(required = false)
private ProxyConsumerResolver proxyConsumerResolver;
@@ -261,21 +265,29 @@ public class RocketMQAdminClientImpl implements AdminClient {
@Override
public TopicVO createTopic(String instanceId, TopicVO topic) {
- topic.setInstanceId(instanceId);
+ return createTopic(instanceId, topic, false);
+ }
+
+ @Override
+ public TopicVO importTopic(String instanceId, TopicVO topic) {
+ return createTopic(instanceId, topic, true);
+ }
+
+ private TopicVO createTopic(String instanceId, TopicVO topic, boolean importing) {
+ String target = ownershipGuard.requireInstance(instanceId).getName();
+ if (topic == null) {
+ throw new BusinessException(400, "Topic request is required");
+ }
+ topic.setInstanceId(target);
+ topic.setName(ResourceOwnershipGuard.requireText(topic.getName(), "topicName"));
String topicName = topic.getName();
int writeQueues = topic.getWriteQueues() > 0 ? topic.getWriteQueues() : 8;
int readQueues = topic.getReadQueues() > 0 ? topic.getReadQueues() : 8;
- return executeForInstance(instanceId, admin -> {
+ return executeResourceWrite(target, Kind.TOPIC, topicName, true, topic.getType(), (admin, clusterName) -> {
try {
- String clusterName = getClusterName(admin);
- // Match on the instance-scoped identity: the same topic name can exist in several
- // clusters or Studio instances, and a broader lookup would corrupt metadata.
- RmqTopic existing = topicMapper.selectOne(
- new LambdaQueryWrapper<RmqTopic>()
- .eq(RmqTopic::getClusterId, clusterName)
- .eq(RmqTopic::getInstanceId, metadataScope(instanceId))
- .eq(RmqTopic::getName, topicName));
+ RmqTopic existing = topicMapper.selectOne(new LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getName, topicName));
TopicPerm effectivePerm = topic.getPerm() != null
? topic.getPerm()
: existing == null ? TopicPerm.RW : fromRocketMQPerm(existing.getPerm());
@@ -291,26 +303,24 @@ public class RocketMQAdminClientImpl implements AdminClient {
topicConfig.setPerm(toRocketMQPerm(effectivePerm));
applyTopicType(topicConfig, topic.getType());
- for (String addr : brokerAddrs) {
- admin.createAndUpdateTopicConfig(addr, topicConfig);
+ boolean physicalExists = importing && verifyTopicImport(admin, brokerAddrs, topicConfig);
+ if (!physicalExists) {
+ for (String addr : brokerAddrs) {
+ admin.createAndUpdateTopicConfig(addr, topicConfig);
+ }
}
- // Persist to DB. Re-creating a topic that already has a record (for example when
- // rebuilding a broker route from the console) must update it instead of failing on
- // the instance-scoped unique key.
+ // 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::getClusterId, clusterName)
- .eq(RmqTopic::getInstanceId, metadataScope(instanceId))
.eq(RmqTopic::getName, topicName));
boolean isNew = entity == null;
if (isNew) {
entity = new RmqTopic();
entity.setName(topicName);
entity.setClusterId(clusterName);
- entity.setInstanceId(metadataScope(instanceId));
+ entity.setInstanceId(target);
entity.setGmtCreate(LocalDateTime.now());
}
- entity.setInstanceId(metadataScope(instanceId));
if (topic.getType() != null) {
entity.setTopicType(topic.getType().name());
} else if (isNew) {
@@ -349,19 +359,15 @@ public class RocketMQAdminClientImpl implements AdminClient {
@Override
public TopicVO updateTopic(String instanceId, TopicVO topic) {
- topic.setInstanceId(instanceId);
+ String target = ownershipGuard.requireInstance(instanceId).getName();
+ topic.setInstanceId(target);
+ topic.setName(ResourceOwnershipGuard.requireText(topic.getName(), "topicName"));
String topicName = topic.getName();
- return executeForInstance(instanceId, admin -> {
+ return executeResourceWrite(target, Kind.TOPIC, topicName, false, topic.getType(), (admin, clusterName) -> {
try {
- // Match on the instance-scoped identity so same-name resources under different
- // Studio instances do not overwrite each other.
- String clusterName = getClusterName(admin);
- RmqTopic existing = topicMapper.selectOne(
- new LambdaQueryWrapper<RmqTopic>()
- .eq(RmqTopic::getClusterId, clusterName)
- .eq(RmqTopic::getInstanceId, metadataScope(instanceId))
- .eq(RmqTopic::getName, topicName));
+ RmqTopic existing = topicMapper.selectOne(new LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getName, topicName));
if (existing == null) {
throw new BusinessException(404, "Topic not found: " + topicName);
}
@@ -452,34 +458,25 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
@Override
- public void deleteTopic(String instanceId, String name) {
- String namesrvAddr = namesrvAddr(instanceId);
- executeForInstance(instanceId, admin -> {
+ public void deleteTopic(String instanceId, String rawName) {
+ String name = ResourceOwnershipGuard.requireText(rawName, "topicName");
+ InstanceVO instance = ownershipGuard.requireInstance(instanceId);
+ 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 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) -> {
try {
- String clusterName = getClusterName(admin);
Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin, clusterName);
- // Delete from brokers
- if (!brokerAddrs.isEmpty()) {
- admin.deleteTopicInBroker(brokerAddrs, name);
- }
-
- // Delete from nameserver
- Set<String> nsAddrs = new HashSet<>();
- for (String addr : namesrvAddr.split("[;,]")) {
- String trimmed = addr.trim();
- if (!trimmed.isEmpty()) {
- nsAddrs.add(trimmed);
- }
- }
- Set<String> clusters = Set.of(clusterName);
- for (String physicalCluster : clusters) admin.deleteTopicInNameServer(nsAddrs, physicalCluster, name);
-
- // Topic names may be shared by several clusters managed by this Studio instance.
- topicMapper.delete(new LambdaQueryWrapper<RmqTopic>()
- .in(RmqTopic::getClusterId, clusters)
- .eq(RmqTopic::getInstanceId, metadataScope(instanceId))
- .eq(RmqTopic::getName, name));
+ deletePhysicalTopic(admin, instance.getName(), name, clusterName, brokerAddrs);
+ topicMapper.delete(new LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName, name));
recordAudit("DELETE_TOPIC", name, "", "SUCCESS");
return null;
@@ -495,18 +492,33 @@ public class RocketMQAdminClientImpl implements AdminClient {
@Override
public SendMessageVO sendMessage(SendMessageDTO request) {
+ if (request == null) {
+ throw new BusinessException(400, "Send message request is required");
+ }
+ ResourceOwnershipGuard.requireText(request.getInstanceId(), "instanceId");
+ validateMessageSize(request);
+ InstanceVO instance = ownershipGuard.requireInstance(request.getInstanceId());
+ request.setInstanceId(instance.getName());
+ request.setTopic(ResourceOwnershipGuard.requireText(request.getTopic(), "topicName"));
+ Resource resource = ownershipGuard.topicResource(request.getTopic());
+ ownershipGuard.check(instance, resource, true);
+ return ownershipGuard.withOwned(instance, List.of(resource), () -> {
+ executeForInstance(instance.getName(), admin -> {
+ var owner = ownershipGuard.check(instance, resource, true);
+ var target = ApacheWriteTargetResolver.resolve(admin, instance, owner.clusterId());
+ ApacheWriteTargetResolver.requireTopicRoute(admin, target, request.getTopic());
+ return null;
+ });
+ return sendOwnedMessage(request);
+ });
+ }
+
+ private SendMessageVO sendOwnedMessage(SendMessageDTO request) {
String topic = request.getTopic();
String tag = request.getTag() != null ? request.getTag() : "";
String key = request.getKey() != null ? request.getKey() : "";
String body = request.getBody() != null ? request.getBody() : "";
byte[] bodyBytes = body.getBytes(StandardCharsets.UTF_8);
- if (bodyBytes.length > MAX_MESSAGE_SIZE) {
- String message = "Message body size " + bodyBytes.length
- + " exceeds the maximum of " + MAX_MESSAGE_SIZE + " bytes";
- recordAudit("SEND_MESSAGE", topic, message, "FAILED");
- throw new BusinessException(400, message);
- }
-
MqClientPool.ClientAction<DefaultMQProducer, SendMessageVO> sendAction = producer -> {
Message msg = new Message(topic, bodyBytes);
if (StringUtils.hasText(tag)) {
@@ -557,9 +569,7 @@ public class RocketMQAdminClientImpl implements AdminClient {
.build();
};
try {
- return StringUtils.hasText(request.getInstanceId())
- ? runtimeAdminClientResolver.executeProducer(request.getInstanceId(), sendAction)
- : clientPool.withProducer(namesrvAddr(), null, null, sendAction);
+ return runtimeAdminClientResolver.executeProducer(request.getInstanceId(), sendAction);
} catch (BusinessException e) {
recordAudit("SEND_MESSAGE", request.getTopic(), e.getMessage(), "FAILED");
throw e;
@@ -569,6 +579,16 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
}
+ private void validateMessageSize(SendMessageDTO request) {
+ int size = request.getBody() == null ? 0 : request.getBody().getBytes(StandardCharsets.UTF_8).length;
+ if (size > MAX_MESSAGE_SIZE) {
+ String message = "Message body size " + size
+ + " exceeds the maximum of " + MAX_MESSAGE_SIZE + " bytes";
+ recordAudit("SEND_MESSAGE", request.getTopic(), message, "FAILED");
+ throw new BusinessException(400, message);
+ }
+ }
+
private static boolean isSystemReservedProperty(String key) {
return key == null
|| MessageConst.STRING_HASH_SET.contains(key)
@@ -578,20 +598,32 @@ public class RocketMQAdminClientImpl implements AdminClient {
@Override
public ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group) {
- String instanceId = group != null ? group.getInstanceId() : null;
- return executeForInstance(instanceId,
- admin -> createConsumerGroup(admin, group));
+ return createConsumerGroup(group, false);
+ }
+
+ @Override
+ public ConsumerGroupVO importConsumerGroup(ConsumerGroupVO group) {
+ return createConsumerGroup(group, true);
+ }
+
+ private ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group, boolean importing) {
+ String instanceId = ownershipGuard.requireInstance(group == null ? null : group.getInstanceId()).getName();
+ 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));
}
@Override
public ConsumerGroupVO updateConsumerGroup(ConsumerGroupVO group) {
- String instanceId = group != null ? group.getInstanceId() : null;
- return executeForInstance(instanceId, admin -> {
+ String instanceId = ownershipGuard.requireInstance(group == null ? null : group.getInstanceId()).getName();
+ group.setInstanceId(instanceId);
+ group.setName(ResourceOwnershipGuard.requireText(group.getName(), "groupName"));
+ return executeResourceWrite(instanceId, Kind.GROUP, group.getName(), false, null, (admin, clusterName) -> {
String groupName = group.getName();
int totalBrokers = 0;
int updatedBrokers = 0;
try {
- String clusterName = getClusterName(admin);
Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin, clusterName);
if (brokerAddrs.isEmpty()) {
throw new BusinessException(502, "No broker available to update consumer group");
@@ -635,7 +667,9 @@ public class RocketMQAdminClientImpl implements AdminClient {
public ConsumerGroupSettingsVO getConsumerGroupSettings(String instanceId, String name) {
return executeForInstance(instanceId, admin -> {
try {
- Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin, getClusterName(admin));
+ InstanceVO instance = ownershipGuard.requireInstance(instanceId);
+ var owner = ownershipGuard.check(instance, new Resource(Kind.GROUP, name), 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");
}
@@ -656,13 +690,13 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
@Override
- public ConsumerGroupSettingsVO updateConsumerGroupSettings(String instanceId, String name,
+ public ConsumerGroupSettingsVO updateConsumerGroupSettings(String instanceId, String rawName,
ConsumerGroupSettingsCommand command) {
- return executeForInstance(instanceId, admin -> {
+ String name = ResourceOwnershipGuard.requireText(rawName, "groupName");
+ return executeResourceWrite(instanceId, Kind.GROUP, name, false, null, (admin, clusterName) -> {
int totalBrokers = 0;
int updatedBrokers = 0;
try {
- String clusterName = getClusterName(admin);
Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin, clusterName);
if (brokerAddrs.isEmpty()) {
throw new BusinessException(502, "No broker available to update consumer group settings");
@@ -698,8 +732,6 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
}
RmqGroup group = groupMapper.selectOne(new LambdaQueryWrapper<RmqGroup>()
- .eq(RmqGroup::getClusterId, clusterName)
- .eq(RmqGroup::getInstanceId, metadataScope(instanceId))
.eq(RmqGroup::getName, name));
if (group != null) {
group.setMaxRetry(command.retryMaxTimes());
@@ -735,11 +767,11 @@ public class RocketMQAdminClientImpl implements AdminClient {
return value == null ? "preserved" : value.toString();
}
- private ConsumerGroupVO createConsumerGroup(MQAdminExt admin, ConsumerGroupVO group) {
+ private ConsumerGroupVO createConsumerGroup(MQAdminExt admin, ConsumerGroupVO group, String groupClusterName,
+ boolean importing) {
String groupName = group.getName();
try {
- String groupClusterName = getClusterName(admin);
Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin, groupClusterName);
if (brokerAddrs.isEmpty()) {
throw new BusinessException(500, "No broker available to create consumer group");
@@ -752,8 +784,11 @@ public class RocketMQAdminClientImpl implements AdminClient {
config.setRetryQueueNums(1);
config.setRetryMaxTimes(group.getRetryMaxTimes() > 0 ? group.getRetryMaxTimes() : 16);
- for (String addr : brokerAddrs) {
- admin.createAndUpdateSubscriptionGroupConfig(addr, config);
+ boolean physicalExists = importing && verifyGroupImport(admin, brokerAddrs, config);
+ if (!physicalExists) {
+ for (String addr : brokerAddrs) {
+ admin.createAndUpdateSubscriptionGroupConfig(addr, config);
+ }
}
persistConsumerGroup(group, groupClusterName, config.getRetryMaxTimes());
@@ -770,21 +805,17 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
private void persistConsumerGroup(ConsumerGroupVO group, String clusterName, int retryMaxTimes) {
- // Only metadata persistence is shared by create and update; broker writes have
- // separate semantics. Scope the upsert to the selected Studio instance.
+ // Ownership is determined by the global reservation guard; a three-column filter must not mask cross-instance conflicts.
RmqGroup entity = groupMapper.selectOne(new LambdaQueryWrapper<RmqGroup>()
- .eq(RmqGroup::getClusterId, clusterName)
- .eq(RmqGroup::getInstanceId, metadataScope(group.getInstanceId()))
.eq(RmqGroup::getName, group.getName()));
boolean isNewGroup = entity == null;
if (isNewGroup) {
entity = new RmqGroup();
entity.setName(group.getName());
entity.setClusterId(clusterName);
- entity.setInstanceId(metadataScope(group.getInstanceId()));
+ entity.setInstanceId(group.getInstanceId());
entity.setGmtCreate(LocalDateTime.now());
}
- entity.setInstanceId(metadataScope(group.getInstanceId()));
entity.setConsumeType(group.getConsumeType() != null ? group.getConsumeType().name() : "CLUSTERING");
entity.setMessageModel(group.getSubscriptionMode() != null ? group.getSubscriptionMode().name() : "Push");
entity.setMaxRetry(retryMaxTimes);
@@ -799,27 +830,28 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
@Override
- public void deleteConsumerGroup(String instanceId, String name) {
- executeForInstance(instanceId, admin -> {
- doDeleteConsumerGroup(instanceId, admin, name);
+ public void deleteConsumerGroup(String instanceId, String rawName) {
+ String target = ownershipGuard.requireInstance(instanceId).getName();
+ String name = ResourceOwnershipGuard.requireText(rawName, "groupName");
+ executeResourceWrite(target, Kind.GROUP, name, false, null, (admin, clusterName) -> {
+ doDeleteConsumerGroup(target, admin, name, clusterName);
return null;
});
}
- private void doDeleteConsumerGroup(String instanceId, MQAdminExt admin, String name) {
+ private void doDeleteConsumerGroup(String instanceId, MQAdminExt admin, String name, String clusterName) {
try {
- String clusterName = getClusterName(admin);
Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin, clusterName);
+ resolveNameservers(instanceId);
for (String addr : brokerAddrs) {
admin.deleteSubscriptionGroup(addr, name, true);
}
- // Consumer group names may be shared by several clusters managed by this Studio instance.
- groupMapper.delete(new LambdaQueryWrapper<RmqGroup>()
- .eq(RmqGroup::getClusterId, clusterName)
- .eq(RmqGroup::getInstanceId, metadataScope(instanceId))
- .eq(RmqGroup::getName, name));
+ // 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));
recordAudit("DELETE_GROUP", name, "", "SUCCESS");
} catch (BusinessException e) {
@@ -1073,27 +1105,35 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
@Override
- public void resetOffset(String instanceId, String name, long timestamp, String topic) {
- if (!StringUtils.hasText(topic)) {
+ public void resetOffset(String instanceId, String rawName, long timestamp, String rawTopic) {
+ if (!StringUtils.hasText(rawTopic)) {
throw new BusinessException(400, "topic is required for offset reset");
}
+ String name = ResourceOwnershipGuard.requireText(rawName, "groupName");
+ String topic = rawTopic.trim();
try {
// resetOffsetNew applies the searched offset even when it is ahead of the
// current offset (the preview offers "skip unconsumed messages") and falls
// back to direct offset writes when the group is offline; the non-forcing
// resetOffsetByTimestamp silently capped forward targets to the current
// offset and failed with CONSUMER_NOT_ONLINE for offline groups.
- if (StringUtils.hasText(instanceId)) {
- runtimeAdminClientResolver.execute(instanceId, admin -> {
- admin.resetOffsetNew(name, topic, timestamp);
- return null;
- });
- } else {
- adminFactory.execute(namesrvAddr(), null, admin -> {
- admin.resetOffsetNew(name, topic, timestamp);
- return null;
- });
- }
+ 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.withOwned(instance, List.of(group, topicResource), () ->
+ executeForInstance(instance.getName(), admin -> {
+ var groupOwner = ownershipGuard.check(instance, group, true);
+ var topicOwner = ownershipGuard.check(instance, topicResource, true);
+ if (!groupOwner.clusterId().equals(topicOwner.clusterId())) {
+ throw new BusinessException(409, "Group and topic are not in the same target cluster");
+ }
+ var target = ApacheWriteTargetResolver.resolve(admin, instance, groupOwner.clusterId());
+ ApacheWriteTargetResolver.requireTopicRoute(admin, target, topic);
+ admin.resetOffsetNew(name, topic, timestamp);
+ return null;
+ }));
recordAudit("RESET_OFFSET", name,
"instanceId=" + instanceId + ", topic=" + topic + ", timestamp=" + timestamp, "SUCCESS");
} catch (BusinessException e) {
@@ -1156,87 +1196,110 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
private String namesrvAddr(String instanceId) {
- return StringUtils.hasText(instanceId)
- ? runtimeAdminClientResolver.resolveEndpoint(instanceId)
- : namesrvAddr();
+ return runtimeAdminClientResolver.resolveEndpoint(ResourceOwnershipGuard.requireText(instanceId, "instanceId"));
}
private <T> T executeForInstance(String instanceId, MqAdminExtFactory.AdminAction<T> action) {
- if (StringUtils.hasText(instanceId)) {
- return runtimeAdminClientResolver.execute(instanceId, action);
- }
- return adminFactory.execute(namesrvAddr(), null, action);
+ return runtimeAdminClientResolver.execute(ResourceOwnershipGuard.requireText(instanceId, "instanceId"), action);
}
- private String metadataScope(String instanceId) {
- if (StringUtils.hasText(instanceId) && runtimeAdminClientResolver.configuredClusterName(instanceId) != null) {
- return LEGACY_METADATA_SCOPE;
+ private <T> T executeResourceWrite(String instanceId, Kind kind, String name, boolean create, TopicType type,
+ 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) {
+ throw new BusinessException(400, "Derived resources cannot be registered as a regular topic");
}
- return StringUtils.hasText(instanceId) ? instanceId.trim() : LEGACY_METADATA_SCOPE;
+ var owner = ownershipGuard.check(instance, resource, !create);
+ 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()));
+ });
}
- private Set<String> getMasterBrokerAddrsForCluster(MQAdminExt admin, String clusterName) throws Exception {
- if (!StringUtils.hasText(clusterName)) {
- return getAllMasterBrokerAddrs(admin);
- }
- ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
- if (clusterInfo == null || clusterInfo.getClusterAddrTable() == null
- || clusterInfo.getBrokerAddrTable() == null) {
- return getAllMasterBrokerAddrs(admin);
- }
- Set<String> brokerNames = clusterInfo.getClusterAddrTable().get(clusterName);
- if (brokerNames == null || brokerNames.isEmpty()) {
- return getAllMasterBrokerAddrs(admin);
- }
- Set<String> addrs = new HashSet<>();
- for (String brokerName : brokerNames) {
- BrokerData brokerData = clusterInfo.getBrokerAddrTable().get(brokerName);
- if (brokerData == null || brokerData.getBrokerAddrs() == null) {
- continue;
+ /** 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;
+ for (String master : masters) {
+ TopicConfig current;
+ try {
+ current = admin.examineTopicConfig(master, requested.getTopicName());
+ } catch (Exception failure) {
+ if (MqResponseCodes.hasResponseCode(failure, ResponseCode.TOPIC_NOT_EXIST)) {
+ continue;
+ }
+ throw failure;
}
- String masterAddr = brokerData.getBrokerAddrs().get(0L);
- if (masterAddr == null && !brokerData.getBrokerAddrs().isEmpty()) {
- masterAddr = brokerData.getBrokerAddrs().values().iterator().next();
+ if (current == null) {
+ continue;
}
- if (masterAddr != null) {
- addrs.add(masterAddr);
+ found = true;
+ if (!requested.getTopicName().equals(current.getTopicName())
+ || requested.getReadQueueNums() != current.getReadQueueNums()
+ || requested.getWriteQueueNums() != current.getWriteQueueNums()
+ || requested.getPerm() != current.getPerm()
+ // The write protocol's +message.type is not the bare key used when reading config; do not read the request's type default directly.
+ || !requested.getAttributes().getOrDefault("+" + TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(),
+ requested.getTopicMessageType().name()).equals(current.getTopicMessageType().name())) {
+ throw new BusinessException(409, "Import config conflicts with the existing physical topic; refusing to overwrite: " + requested.getTopicName());
}
}
- return addrs.isEmpty() ? getAllMasterBrokerAddrs(admin) : addrs;
+ return found;
}
- private Set<String> getAllMasterBrokerAddrs(MQAdminExt admin) throws Exception {
- Set<String> addrs = new HashSet<>();
- ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
- if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
- return addrs;
- }
-
- for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {
- if (brokerData == null || brokerData.getBrokerAddrs() == null) {
- continue;
+ private boolean verifyGroupImport(MQAdminExt admin, Set<String> masters, SubscriptionGroupConfig requested)
+ throws Exception {
+ boolean found = false;
+ for (String master : masters) {
+ SubscriptionGroupConfig current;
+ try {
+ current = admin.examineSubscriptionGroupConfig(master, requested.getGroupName());
+ } catch (Exception failure) {
+ if (MqResponseCodes.hasResponseCode(failure, ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST)) {
+ continue;
+ }
+ throw failure;
}
- String masterAddr = brokerData.getBrokerAddrs().get(0L);
- if (masterAddr == null && !brokerData.getBrokerAddrs().isEmpty()) {
- masterAddr = brokerData.getBrokerAddrs().values().iterator().next();
+ if (current == null) {
+ continue;
}
- if (masterAddr != null) {
- addrs.add(masterAddr);
+ found = true;
+ if (!requested.getGroupName().equals(current.getGroupName())
+ || requested.getRetryMaxTimes() != current.getRetryMaxTimes()) {
+ throw new BusinessException(409, "Import config conflicts with the existing physical group; refusing to overwrite: " + requested.getGroupName());
}
}
- return addrs;
+ return found;
}
- private String getClusterName(MQAdminExt admin) {
- try {
- ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
- if (clusterInfo != null && clusterInfo.getClusterAddrTable() != null
- && !clusterInfo.getClusterAddrTable().isEmpty()) {
- return clusterInfo.getClusterAddrTable().keySet().iterator().next();
+ private Set<String> getMasterBrokerAddrsForCluster(MQAdminExt admin, String clusterName) throws Exception {
+ return ApacheWriteTargetResolver.target(admin.examineBrokerClusterInfo(), clusterName).masters();
+ }
+
+ private void deletePhysicalTopic(MQAdminExt admin, String instanceId, String name, String cluster,
+ Set<String> brokers) throws Exception {
+ Set<String> nameservers = resolveNameservers(instanceId);
+ admin.deleteTopicInBroker(brokers, name);
+ admin.deleteTopicInNameServer(nameservers, cluster, name);
+ }
+
+ private Set<String> resolveNameservers(String instanceId) {
+ Set<String> nameservers = new HashSet<>();
+ String endpoint = namesrvAddr(instanceId);
+ if (!StringUtils.hasText(endpoint)) {
+ throw new BusinessException(409, "Target instance has no available NameServer");
+ }
+ for (String address : endpoint.split("[;,]")) {
+ if (StringUtils.hasText(address)) {
+ nameservers.add(address.trim());
}
- } catch (Exception ignored) {
}
- return "DefaultCluster";
+ if (nameservers.isEmpty()) {
+ throw new BusinessException(409, "Target instance has no available NameServer");
+ }
+ return nameservers;
}
private int toRocketMQPerm(TopicPerm perm) {
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 16fdd5ceb..cfae98e34 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
@@ -32,6 +32,8 @@ import org.apache.rocketmq.remoting.protocol.body.GetLiteGroupInfoResponseBody;
import org.apache.rocketmq.remoting.protocol.body.GetParentTopicInfoResponseBody;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.model.LiteTopicQuota;
import org.apache.rocketmq.studio.model.LiteTopicSession;
@@ -100,6 +102,8 @@ public class RocketMQLiteTopicProvider implements LiteTopicProvider {
private final MqAdminExtFactory adminFactory;
private final RocketMQProperties properties;
+ private final RuntimeAdminClientResolver runtimeAdminClientResolver;
+ private final ResourceOwnershipGuard ownershipGuard;
// โโโ Capability โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
@@ -331,22 +335,28 @@ public class RocketMQLiteTopicProvider implements LiteTopicProvider {
// โโโ TTL update โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
@Override
- public void extendTTL(String topicPattern, long ttlMillis) {
- requireAdmin();
- if (!StringUtils.hasText(topicPattern)) {
- throw new BusinessException(400, "topicPattern is required");
+ public void extendTTL(String instanceId, String rawTopicPattern, long ttlMillis) {
+ var instance = ownershipGuard.requireInstance(instanceId);
+ String topicPattern = ResourceOwnershipGuard.requireText(rawTopicPattern, "topicPattern");
+ var resource = ownershipGuard.topicResource(topicPattern);
+ ownershipGuard.check(instance, resource, 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");
}
if (ttlMillis <= 0) {
throw new BusinessException(400, "newTTL must be positive");
}
long minutes = Math.min(Math.max(Math.round(ttlMillis / 60000.0), 1), MAX_LITE_TTL_MINUTES);
- execute(admin -> {
+ ownershipGuard.withOwned(instance, List.of(resource), () -> runtimeAdminClientResolver.execute(instance, admin -> {
+ var owner = ownershipGuard.check(instance, resource, 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
// would leave the cluster with mixed lite.topic.expiration attributes while the
// console reports a fully successful extension.
Map<String, TopicConfig> pending = new LinkedHashMap<>();
- for (String master : masterAddresses(admin)) {
+ for (String master : target.masters().stream().sorted().toList()) {
TopicConfig config = liteParentTopicConfig(admin, master, topicPattern);
if (config != null) {
pending.put(master, config);
@@ -370,7 +380,7 @@ public class RocketMQLiteTopicProvider implements LiteTopicProvider {
log.info("Extended LiteTopic TTL to {}ms ({} min) for parent topic {} on {} broker(s)",
ttlMillis, minutes, topicPattern, pending.size());
return null;
- });
+ }));
}
/**
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 d17bb88ec..3405d1ce2 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
@@ -31,6 +31,9 @@ import org.apache.rocketmq.remoting.protocol.route.QueueData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
import org.apache.rocketmq.studio.common.util.MessagePropertyDisplay;
import org.apache.rocketmq.studio.common.util.MqResponseCodes;
import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
@@ -100,6 +103,7 @@ public class RocketMQMessageProvider implements MessageProvider {
.thenComparing(MessageRecordVO::getMsgId, Comparator.nullsFirst(String::compareTo));
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
+ private final ResourceOwnershipGuard ownershipGuard;
@Override
public List<MessageRecordVO> queryMessages(String instanceId, String topic, String msgId, String tag, String key,
@@ -538,15 +542,63 @@ public class RocketMQMessageProvider implements MessageProvider {
@Override
public DirectConsumeMessageResultVO consumeMessageDirectly(DirectConsumeMessageDTO request) {
- return runtimeAdminClientResolver.execute(request.getInstanceId(), admin -> {
- org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult result =
- ((DefaultMQAdminExt) admin).consumeMessageDirectly(request.getConsumerGroup(), request.getClientId(),
- request.getTopic(), request.getMsgId());
- return DirectConsumeMessageResultVO.builder()
- .consumeResult(result.getConsumeResult() == null ? "UNKNOWN" : result.getConsumeResult().name())
- .remark(result.getRemark()).spentTimeMillis(result.getSpentTimeMills())
- .order(result.isOrder()).autoCommit(result.isAutoCommit()).build();
- });
+ if (request == null) {
+ throw new BusinessException(400, "Direct consume request is required");
+ }
+ var instance = ownershipGuard.requireInstance(request.getInstanceId());
+ String topic = ResourceOwnershipGuard.requireText(request.getTopic(), "topicName");
+ String group = ResourceOwnershipGuard.requireText(request.getConsumerGroup(), "groupName");
+ String client = ResourceOwnershipGuard.requireText(request.getClientId(), "clientId");
+ 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);
+ 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);
+ if (!topicOwner.clusterId().equals(groupOwner.clusterId())) {
+ throw new BusinessException(409, "Group and topic are not in the same target cluster");
+ }
+ var target = ApacheWriteTargetResolver.resolve(admin, instance, topicOwner.clusterId());
+ ApacheWriteTargetResolver.requireTopicRoute(admin, target, topic);
+ // offsetId can bypass topic routing and connect directly to a broker; verify the address first, then the message's actual ownership.
+ requireDirectMessageTarget(target, messageId, false);
+ MessageExt message = ((DefaultMQAdminExt) admin).viewMessage(topic, messageId);
+ if (message == null || !topic.equals(message.getTopic())) {
+ throw new BusinessException(409, "Message's actual topic does not match the authorized resource");
+ }
+ requireDirectMessageTarget(target, message.getMsgId(), true);
+ if (!(message.getStoreHost() instanceof java.net.InetSocketAddress host)
+ || host.getAddress() == null
+ || !target.masters().contains(host.getAddress().getHostAddress() + ":" + host.getPort())) {
+ throw new BusinessException(409, "Message's store broker is not in the target master set");
+ }
+ var result = ((DefaultMQAdminExt) admin).consumeMessageDirectly(group, client,
+ topic, message.getMsgId());
+ return DirectConsumeMessageResultVO.builder()
+ .consumeResult(result.getConsumeResult() == null ? "UNKNOWN" : result.getConsumeResult().name())
+ .remark(result.getRemark()).spentTimeMillis(result.getSpentTimeMills())
+ .order(result.isOrder()).autoCommit(result.isAutoCommit()).build();
+ }));
+ }
+
+ private void requireDirectMessageTarget(ApacheWriteTargetResolver.Target target, String messageId,
+ boolean offsetRequired) {
+ MessageId decoded;
+ try {
+ decoded = MessageDecoder.decodeMessageId(messageId);
+ } catch (Exception invalid) {
+ if (!offsetRequired) {
+ return;
+ }
+ throw new BusinessException(409, "Unable to determine the physical location of the message");
+ }
+ String address = BrokerTopologyGuards.decodedBrokerAddr(decoded);
+ if (address == null || !target.masters().contains(address)) {
+ throw new BusinessException(409, "Message ID does not belong to a target cluster master");
+ }
}
@Override
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index 1240d28a6..7f43cf5a1 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -21,11 +21,9 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
-import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
@@ -145,9 +143,14 @@ public class RocketMQMetadataProvider implements MetadataProvider {
String configuredCluster = StringUtils.hasText(instanceId)
? runtimeAdminClientResolver.configuredClusterName(instanceId) : null;
LambdaQueryWrapper<RmqTopic> query = new LambdaQueryWrapper<RmqTopic>()
- .eq(instanceId != null, RmqTopic::getInstanceId,
- configuredCluster == null ? normalizeMetadataScope(instanceId) : "")
- .eq(configuredCluster != null, RmqTopic::getClusterId, configuredCluster)
+ .and(instanceId != null, scope -> {
+ scope.eq(RmqTopic::getInstanceId, normalizeMetadataScope(instanceId));
+ if (configuredCluster != null) {
+ scope.or(legacy -> legacy.eq(RmqTopic::getClusterId, configuredCluster)
+ .and(empty -> empty.eq(RmqTopic::getInstanceId, "")
+ .or().isNull(RmqTopic::getInstanceId)));
+ }
+ })
.eq(StringUtils.hasText(clusterId), RmqTopic::getClusterId, clusterId)
.eq(StringUtils.hasText(type), RmqTopic::getTopicType, type)
.like(StringUtils.hasText(search), RmqTopic::getName, search)
@@ -169,9 +172,14 @@ public class RocketMQMetadataProvider implements MetadataProvider {
String configuredCluster = StringUtils.hasText(instanceId)
? runtimeAdminClientResolver.configuredClusterName(instanceId) : null;
LambdaQueryWrapper<RmqTopic> query = new LambdaQueryWrapper<RmqTopic>()
- .eq(instanceId != null, RmqTopic::getInstanceId,
- configuredCluster == null ? normalizeMetadataScope(instanceId) : "")
- .eq(configuredCluster != null, RmqTopic::getClusterId, configuredCluster)
+ .and(instanceId != null, scope -> {
+ scope.eq(RmqTopic::getInstanceId, normalizeMetadataScope(instanceId));
+ if (configuredCluster != null) {
+ scope.or(legacy -> legacy.eq(RmqTopic::getClusterId, configuredCluster)
+ .and(empty -> empty.eq(RmqTopic::getInstanceId, "")
+ .or().isNull(RmqTopic::getInstanceId)));
+ }
+ })
.eq(StringUtils.hasText(clusterId), RmqTopic::getClusterId, clusterId)
.eq(StringUtils.hasText(type), RmqTopic::getTopicType, type)
.like(StringUtils.hasText(search), RmqTopic::getName, search)
@@ -234,9 +242,14 @@ public class RocketMQMetadataProvider implements MetadataProvider {
String configuredCluster = StringUtils.hasText(instanceId)
? runtimeAdminClientResolver.configuredClusterName(instanceId) : null;
LambdaQueryWrapper<RmqGroup> query = new LambdaQueryWrapper<RmqGroup>()
- .eq(instanceId != null, RmqGroup::getInstanceId,
- configuredCluster == null ? normalizeMetadataScope(instanceId) : "")
- .eq(configuredCluster != null, RmqGroup::getClusterId, configuredCluster)
+ .and(instanceId != null, scope -> {
+ scope.eq(RmqGroup::getInstanceId, normalizeMetadataScope(instanceId));
+ if (configuredCluster != null) {
+ scope.or(legacy -> legacy.eq(RmqGroup::getClusterId, configuredCluster)
+ .and(empty -> empty.eq(RmqGroup::getInstanceId, "")
+ .or().isNull(RmqGroup::getInstanceId)));
+ }
+ })
.eq(StringUtils.hasText(clusterId), RmqGroup::getClusterId, clusterId)
.like(StringUtils.hasText(search), RmqGroup::getName, search)
.orderByAsc(RmqGroup::getName);
@@ -255,9 +268,14 @@ public class RocketMQMetadataProvider implements MetadataProvider {
String configuredCluster = StringUtils.hasText(instanceId)
? runtimeAdminClientResolver.configuredClusterName(instanceId) : null;
LambdaQueryWrapper<RmqGroup> query = new LambdaQueryWrapper<RmqGroup>()
- .eq(instanceId != null, RmqGroup::getInstanceId,
- configuredCluster == null ? normalizeMetadataScope(instanceId) : "")
- .eq(configuredCluster != null, RmqGroup::getClusterId, configuredCluster)
+ .and(instanceId != null, scope -> {
+ scope.eq(RmqGroup::getInstanceId, normalizeMetadataScope(instanceId));
+ if (configuredCluster != null) {
+ scope.or(legacy -> legacy.eq(RmqGroup::getClusterId, configuredCluster)
+ .and(empty -> empty.eq(RmqGroup::getInstanceId, "")
+ .or().isNull(RmqGroup::getInstanceId)));
+ }
+ })
.eq(StringUtils.hasText(clusterId), RmqGroup::getClusterId, clusterId)
.like(StringUtils.hasText(search), RmqGroup::getName, search)
.orderByAsc(RmqGroup::getName, RmqGroup::getId);
@@ -635,7 +653,7 @@ public class RocketMQMetadataProvider implements MetadataProvider {
private List<QueueProgressVO> getGroupProgress(MQAdminExt admin, String name) {
try {
- ensureRetryTopicExists(admin, name);
+ // Reads do not create retry topics; a missing route is treated as an empty business state.
ConsumeStats stats = admin.examineConsumeStats(name);
if (stats == null || stats.getOffsetTable() == null) {
return Collections.emptyList();
@@ -688,7 +706,7 @@ public class RocketMQMetadataProvider implements MetadataProvider {
private List<SubscriptionEntryVO> getGroupSubscriptions(MQAdminExt admin, String instanceId, String name) {
try {
- ensureRetryTopicExists(admin, name);
+ // Reads do not create retry topics; a missing route is treated as an empty business state.
ConsumerConnection conn = admin.examineConsumerConnectionInfo(name);
if (conn == null || conn.getSubscriptionTable() == null) {
return Collections.emptyList();
@@ -744,6 +762,9 @@ public class RocketMQMetadataProvider implements MetadataProvider {
}
private boolean isGroupNotOnline(Exception e) {
+ if (MqResponseCodes.hasResponseCode(e, ResponseCode.TOPIC_NOT_EXIST)) {
+ return true;
+ }
if (e instanceof org.apache.rocketmq.client.exception.MQBrokerException brokerException) {
return brokerException.getResponseCode()
== org.apache.rocketmq.remoting.protocol.ResponseCode.CONSUMER_NOT_ONLINE;
@@ -766,50 +787,6 @@ public class RocketMQMetadataProvider implements MetadataProvider {
// โโ Helper methods โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
- /**
- * examineConsumeStats / examineConsumerConnectionInfo locate brokers through the
- * {@code %RETRY%<group>} topic route, but 5.x POP consumer groups only get their
- * retry topic created once a retry actually happens, so brand-new groups fail with
- * CODE 17. Create the standard retry topic on every master broker when missing
- * (same shape the broker itself creates lazily); the call is idempotent.
- */
- private void ensureRetryTopicExists(MQAdminExt admin, String groupName) {
- String retryTopic = MixAll.getRetryTopic(groupName);
- try {
- TopicRouteData route = admin.examineTopicRouteInfo(retryTopic);
- if (route != null && route.getBrokerDatas() != null && !route.getBrokerDatas().isEmpty()) {
- return;
- }
- } catch (Exception e) {
- log.info("Retry topic {} not routable yet, creating it: {}", retryTopic, e.getMessage());
- }
- try {
- ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
- if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
- return;
- }
- TopicConfig retryConfig = new TopicConfig();
- retryConfig.setTopicName(retryTopic);
- retryConfig.setReadQueueNums(1);
- retryConfig.setWriteQueueNums(1);
- retryConfig.setPerm(PermName.PERM_READ | PermName.PERM_WRITE);
- for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {
- String brokerAddr = brokerData.selectBrokerAddr();
- if (brokerAddr == null || brokerAddr.isBlank()) {
- continue;
- }
- try {
- admin.createAndUpdateTopicConfig(brokerAddr, retryConfig);
- } catch (Exception e) {
- log.warn("Failed to create retry topic {} on broker {}: {}",
- retryTopic, brokerAddr, e.getMessage());
- }
- }
- } catch (Exception e) {
- log.warn("Failed to ensure retry topic {} exists: {}", retryTopic, e.getMessage());
- }
- }
-
/**
* Resolves the lag for a single queue without clamping the broker's {@code -1} "unknown"
* sentinel to zero. A negative raw diff (typical for RocketMQ 5.0 gRPC consumers) is passed
diff --git a/server/src/main/resources/db/schema.sql b/server/src/main/resources/db/schema.sql
index 4b163e01e..094468060 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -1,3 +1,18 @@
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements. See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
-- server/src/main/resources/db/schema.sql
-- RocketMQ Studio ๆฐๆฎๅบ Schema๏ผMySQL 8.0๏ผ
-- ๆญคๆไปถไธบๅฏไธๆๅจ DDL ๆฅๆบ๏ผMyBatis-Plus Entity ไธๆญคไฟๆๅๆญฅ
@@ -93,6 +108,16 @@ CREATE TABLE IF NOT EXISTS rmq_instance (
REFERENCES rmq_cloud_credential(id) ON DELETE RESTRICT
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+-- ๅฎไพๅ็งฐไบๆฅ่ก๏ผ่ๆๅฎไพๅๅ
ฅไธๅๅๆณจๅใๅ ้คใ่ฟๆฅๅๆดๅ
ฑไบซ้้กบๅบใ
+CREATE TABLE IF NOT EXISTS rmq_instance_ownership_lock (
+ `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'ไธป้ฎ',
+ `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'ๅๅปบๆถ้ด',
+ `gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'ไฟฎๆนๆถ้ด',
+ name VARCHAR(128) NOT NULL,
+ PRIMARY KEY (`id`),
+ UNIQUE KEY uk_ownership_instance_name (name)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
-- 3. Topic ็ฎก็่ฎฐๅฝ๏ผ้่ฟ Studio ๅๅปบ/็ฎก็็ Topic ๅ
ๆฐๆฎ๏ผ
CREATE TABLE IF NOT EXISTS rmq_instance_topic (
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'ไธป้ฎ',
@@ -109,8 +134,9 @@ CREATE TABLE IF NOT EXISTS rmq_instance_topic (
status VARCHAR(32) DEFAULT 'ACTIVE',
created_by VARCHAR(64),
PRIMARY KEY (`id`),
- UNIQUE KEY uk_cluster_instance_topic (cluster_id, instance_id, name),
- INDEX idx_topic_instance (instance_id)
+ UNIQUE KEY uk_topic_name (name),
+ INDEX idx_topic_instance (instance_id),
+ INDEX idx_topic_cluster (cluster_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 4. Consumer Group ็ฎก็่ฎฐๅฝ
@@ -127,8 +153,9 @@ CREATE TABLE IF NOT EXISTS rmq_instance_group (
status VARCHAR(32) DEFAULT 'ACTIVE',
created_by VARCHAR(64),
PRIMARY KEY (`id`),
- UNIQUE KEY uk_cluster_instance_group (cluster_id, instance_id, name),
- INDEX idx_group_instance (instance_id)
+ UNIQUE KEY uk_group_name (name),
+ INDEX idx_group_instance (instance_id),
+ INDEX idx_group_cluster (cluster_id)
) 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 d1b21a0d6..91d6cd3ed 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
@@ -106,9 +106,43 @@ class InstanceServiceTest {
@Mock
private RegionNames regionNames;
+ @Mock
+ private ResourceOwnershipGuard ownershipGuard;
+
@InjectMocks
private InstanceService instanceService;
+ @org.junit.jupiter.api.BeforeEach
+ void registrationGuardFixture() {
+ org.mockito.Mockito.lenient().when(ownershipGuard.withInstanceRegistration(anyString(), any()))
+ .thenAnswer(call -> ((java.util.function.Supplier<?>) call.getArgument(1)).get());
+ }
+
+ @Test
+ void deleteOwnedInstanceStopsBeforeProviderTest() {
+ org.mockito.Mockito.doThrow(new BusinessException(409, "Instance still holds resources"))
+ .when(ownershipGuard).lockForInstanceDeletion(1L);
+ assertThatThrownBy(() -> instanceService.deleteInstance(1L))
+ .isInstanceOfSatisfying(BusinessException.class, ex -> assertThat(ex.getCode()).isEqualTo(409));
+ verifyNoInteractions(instanceRepository, providerRegistry, adminFactory, clientPool);
+ }
+
+ @Test
+ void updateOwnedInstanceStopsBeforeSaveTest() {
+ InstanceVO existing = InstanceVO.builder().name("instance-a")
+ .endpoint("old:9876").vendor(InstanceVendor.APACHE).build();
+ existing.setId(1L);
+ InstanceVO request = InstanceVO.builder().endpoint("new:9876").build();
+ request.setId(1L);
+ when(instanceRepository.findById(1L)).thenReturn(Optional.of(existing));
+ org.mockito.Mockito.doThrow(new BusinessException(409, "Instance still holds resources"))
+ .when(ownershipGuard).lockForInstanceUpdate(eq(existing), any());
+ assertThatThrownBy(() -> instanceService.updateInstance(request))
+ .isInstanceOfSatisfying(BusinessException.class, ex -> assertThat(ex.getCode()).isEqualTo(409));
+ verify(instanceRepository, never()).save(any());
+ verifyNoInteractions(adminFactory, clientPool);
+ }
+
@Test
void listInstancesShouldReturnAllWhenNoFilters() {
List<InstanceVO> instances = List.of(
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
new file mode 100644
index 000000000..0f0407e58
--- /dev/null
+++ b/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
@@ -0,0 +1,685 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.instance;
+
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+import org.apache.rocketmq.studio.provider.apache.RocketMQLiteTopicProvider;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.studio.persistence.entity.RmqGroup;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
+import org.apache.rocketmq.studio.provider.apache.RocketMQDefaultClusterResolver;
+import org.apache.rocketmq.studio.provider.apache.RocketMQAdminClientImpl;
+import org.apache.rocketmq.studio.provider.apache.RocketMQMessageProvider;
+import org.apache.rocketmq.studio.provider.apache.RocketMQProperties;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.MqClientPool;
+import org.apache.rocketmq.studio.ops.audit.AuditService;
+import org.apache.rocketmq.studio.persistence.mapper.RmqTopicMapper;
+import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
+import org.apache.rocketmq.studio.persistence.entity.RmqTopic;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
+import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.message.DirectConsumeMessageDTO;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.QueueData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+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.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+
+class ResourceOwnershipGuardTest {
+ private JdbcDataSource dataSource;
+ private JdbcTemplate jdbc;
+ private DataSourceTransactionManager transactions;
+ private ResourceOwnershipGuard guard;
+ private InstanceResolver resolver;
+ private RocketMQDefaultClusterResolver defaultClusters;
+ private InstanceVO first;
+ private InstanceVO second;
+ private final Resource topic = new Resource(Kind.TOPIC, "orders");
+
+ @BeforeEach
+ void setUp() {
+ dataSource = new JdbcDataSource();
+ dataSource.setURL("jdbc:h2:mem:ownership-" + UUID.randomUUID()
+ + ";MODE=MySQL;DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE;LOCK_TIMEOUT=5000");
+ new ResourceDatabasePopulator(new ClassPathResource("db/schema.sql")).execute(dataSource);
+ jdbc = new JdbcTemplate(dataSource);
+ transactions = new DataSourceTransactionManager(dataSource);
+ 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')");
+ first = instance("first");
+ second = instance("second");
+ }
+
+ @Test
+ void missingInstanceAndCanonicalIdentifierTest() {
+ assertCode(() -> guard.requireInstance(" "), 400);
+ when(resolver.findByIdentifier("1")).thenReturn(Optional.of(first));
+ assertThat(guard.requireInstance(" 1 ").getName()).isEqualTo("first");
+ }
+
+ @Test
+ void sameNameserverDifferentInstanceConflictBeforeRemoteTest() {
+ guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
+ AtomicInteger remote = new AtomicInteger();
+ assertCode(() -> guard.write(second, topic, "cluster", true, "NORMAL", remote::incrementAndGet), 409);
+ assertThat(remote).hasValue(0);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_instance_topic", Integer.class)).isEqualTo(1);
+ }
+
+ @Test
+ void partialFailureKeepsCommittedReservationAndAllowsOwnerRetryTest() {
+ AtomicInteger remote = new AtomicInteger();
+ assertThatThrownBy(() -> guard.write(first, topic, "cluster", true, "FIFO", () -> {
+ remote.incrementAndGet();
+ throw new BusinessException(502, "Second broker failed");
+ })).isInstanceOf(BusinessException.class);
+ assertThat(jdbc.queryForObject("SELECT status FROM rmq_instance_topic", String.class)).isEqualTo("PENDING");
+ assertCode(() -> guard.write(second, topic, "cluster", true, "NORMAL", remote::incrementAndGet), 409);
+ guard.write(first, topic, "cluster", true, "FIFO", remote::incrementAndGet);
+ assertThat(remote).hasValue(2);
+ }
+
+ @Test
+ void concurrentClaimsPermitOnlyOneInstancesRemoteCallsTest() throws Exception {
+ CountDownLatch start = new CountDownLatch(1);
+ AtomicInteger remote = new AtomicInteger();
+ try (var executor = Executors.newFixedThreadPool(2)) {
+ var a = executor.submit(() -> claim(start, first, remote));
+ var b = executor.submit(() -> claim(start, second, remote));
+ start.countDown();
+ assertThat(List.of(a.get(10, TimeUnit.SECONDS), b.get(10, TimeUnit.SECONDS)))
+ .containsExactlyInAnyOrder(200, 409);
+ }
+ assertThat(remote).hasValue(1);
+ }
+
+ @Test
+ void instanceDeletionWaitsForClaimAndCannotDropReservationTest() throws Exception {
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ try (var executor = Executors.newFixedThreadPool(2)) {
+ 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 deletion = executor.submit(() -> {
+ assertCode(() -> new TransactionTemplate(transactions).executeWithoutResult(status -> {
+ guard.lockForInstanceDeletion(first.getId());
+ jdbc.update("DELETE FROM rmq_instance WHERE id = ?", first.getId());
+ }), 409);
+ });
+ release.countDown();
+ write.get(10, TimeUnit.SECONDS);
+ deletion.get(10, TimeUnit.SECONDS);
+ }
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_instance", Integer.class)).isEqualTo(2);
+ }
+
+ @Test
+ void deletePartialFailureDoesNotReleaseOwnerTest() {
+ guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
+ assertThatThrownBy(() -> guard.write(first, topic, "cluster", false, null, () -> {
+ 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);
+ }
+
+ @Test
+ void derivedTopicsUseGroupOwnershipWithoutTopicRegistrationTest() {
+ Resource group = new Resource(Kind.GROUP, "buyers");
+ guard.write(first, group, "cluster", true, null, () -> null);
+ assertThat(guard.topicResource("%DLQ%buyers")).isEqualTo(group);
+ assertThat(guard.topicResource("%RETRY%buyers")).isEqualTo(group);
+ assertCode(() -> guard.withOwned(second, List.of(guard.topicResource("%DLQ%buyers")), () -> null), 409);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_instance_topic", Integer.class)).isZero();
+ }
+
+ @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();
+ }
+
+ @Test
+ void migrationIsIdempotentAndLegacyIndexesAreReplacedTest() throws Exception {
+ legacyIndexes();
+ guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
+ jdbc.execute("DROP TABLE rmq_instance_ownership_lock");
+ var migration = new ResourceOwnershipSchemaMigration(dataSource, defaultClusters);
+ 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);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_instance_topic", Integer.class)).isEqualTo(1);
+ }
+
+ @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");
+ 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");
+ assertThat(jdbc.queryForObject("SELECT instance_id FROM rmq_instance_topic", String.class)).isEmpty();
+ }
+
+ @Test
+ void apacheCrudUsesCommittedOwnershipAndOnlySecondClusterMasterTest() throws Exception {
+ 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));
+ when(broker.examineBrokerClusterInfo()).thenReturn(twoClusters());
+ TopicVO request = new TopicVO();
+ request.setName("orders");
+ admin.createTopic("cluster-b", request);
+
+ assertThat(guard.check(virtual, topic, 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() {
+ 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);
+ var broker = mock(DefaultMQAdminExt.class);
+ var admin = apacheAdmin(runtime, broker);
+ when(resolver.findByIdentifier("second")).thenReturn(Optional.of(second));
+ TopicVO request = new TopicVO();
+ request.setName("orders");
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("buyers");
+ group.setInstanceId("second");
+ assertCode(() -> admin.createTopic("second", request), 409);
+ assertCode(() -> admin.updateTopic("second", request), 409);
+ assertCode(() -> admin.deleteTopic("second", "orders"), 409);
+ assertCode(() -> admin.createConsumerGroup(group), 409);
+ assertCode(() -> admin.updateConsumerGroup(group), 409);
+ assertCode(() -> admin.deleteConsumerGroup("second", "buyers"), 409);
+ assertCode(() -> admin.sendMessage(SendMessageDTO.builder()
+ .instanceId("second").topic("orders").body("hello").build()), 409);
+ assertCode(() -> admin.resetOffset("second", "buyers", 1L, "orders"), 409);
+ DirectConsumeMessageDTO direct = new DirectConsumeMessageDTO();
+ direct.setInstanceId("second");
+ direct.setTopic("orders");
+ direct.setConsumerGroup("buyers");
+ direct.setClientId("client");
+ direct.setMsgId("msg");
+ assertCode(() -> new RocketMQMessageProvider(runtime, guard).consumeMessageDirectly(direct), 409);
+ verifyNoInteractions(runtime, broker);
+ }
+
+ @Test
+ void registeredInstanceRejectsAmbiguousAndIncompleteTopologyTest() throws Exception {
+ var runtime = mock(RuntimeAdminClientResolver.class);
+ var broker = mock(DefaultMQAdminExt.class);
+ var admin = apacheAdmin(runtime, broker);
+ when(resolver.findByIdentifier("first")).thenReturn(Optional.of(first));
+ TopicVO request = new TopicVO();
+ request.setName("orders");
+ ClusterInfo topology = twoClusters();
+ when(broker.examineBrokerClusterInfo()).thenReturn(topology);
+ assertCode(() -> admin.createTopic("first", request), 409);
+ topology.getClusterAddrTable().remove("cluster-b");
+ topology.getBrokerAddrTable().get("broker-a").getBrokerAddrs().remove(0L);
+ assertCode(() -> admin.createTopic("first", request), 409);
+ verify(broker, never()).createAndUpdateTopicConfig(anyString(), any(TopicConfig.class));
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_instance_topic", Integer.class)).isZero();
+ }
+
+ @Test
+ void incompleteClusterTableCannotMasqueradeAsUniqueTopologyTest() throws Exception {
+ var runtime = mock(RuntimeAdminClientResolver.class);
+ var broker = mock(DefaultMQAdminExt.class);
+ var admin = apacheAdmin(runtime, broker);
+ when(resolver.findByIdentifier("first")).thenReturn(Optional.of(first));
+ ClusterInfo topology = twoClusters();
+ topology.getClusterAddrTable().remove("cluster-a");
+ when(broker.examineBrokerClusterInfo()).thenReturn(topology);
+ TopicVO request = new TopicVO();
+ request.setName("orders");
+ assertCode(() -> admin.createTopic("first", request), 409);
+ topology.getBrokerAddrTable().get("broker-a").setCluster("cluster-b");
+ assertCode(() -> admin.createTopic("first", request), 409);
+ verify(broker, never()).createAndUpdateTopicConfig(anyString(), any());
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_instance_topic", Integer.class)).isZero();
+ }
+
+ @Test
+ void offsetResetUsesCanonicalNamesWithRealOwnershipTest() throws Exception {
+ guard.write(first, topic, "cluster-b", true, "NORMAL", () -> null);
+ guard.write(first, new Resource(Kind.GROUP, "buyers"), "cluster-b", true, null, () -> null);
+ var runtime = mock(RuntimeAdminClientResolver.class);
+ var broker = mock(DefaultMQAdminExt.class);
+ var admin = apacheAdmin(runtime, broker);
+ when(resolver.findByIdentifier("first")).thenReturn(Optional.of(first));
+ ClusterInfo topology = twoClusters();
+ when(broker.examineBrokerClusterInfo()).thenReturn(topology);
+ TopicRouteData route = new TopicRouteData();
+ route.setBrokerDatas(List.of(topology.getBrokerAddrTable().get("broker-b")));
+ QueueData queue = new QueueData();
+ queue.setBrokerName("broker-b");
+ route.setQueueDatas(List.of(queue));
+ when(broker.examineTopicRouteInfo("orders")).thenReturn(route);
+ admin.resetOffset(" first ", " buyers ", 123L, " orders ");
+ verify(broker).resetOffsetNew("buyers", "orders", 123L);
+ }
+
+ @Test
+ void occupiedInstanceCannotChangeConnectionButMayRotateCredentialTest() {
+ guard.write(first, topic, "cluster-a", true, "NORMAL", () -> null);
+ InstanceVO updated = instance("first");
+ updated.setEndpoint("other:9876");
+ assertCode(() -> new TransactionTemplate(transactions).executeWithoutResult(status ->
+ guard.lockForInstanceUpdate(first, updated)), 409);
+ updated.setEndpoint(first.getEndpoint());
+ updated.setAdminCredentialRef("rotated");
+ new TransactionTemplate(transactions).executeWithoutResult(status -> guard.lockForInstanceUpdate(first, updated));
+ }
+
+ @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");
+ assertThat(jdbc.queryForObject("SELECT instance_id FROM rmq_instance_topic", String.class)).isEqualTo("gone");
+ }
+
+ @Test
+ void importExistingTopicRegistersWithoutOverwritingAndRejectsDifferencesTest() throws Exception {
+ var runtime = mock(RuntimeAdminClientResolver.class);
+ var broker = mock(DefaultMQAdminExt.class);
+ var admin = apacheAdmin(runtime, broker);
+ when(resolver.findByIdentifier("first")).thenReturn(Optional.of(first));
+ ClusterInfo topology = twoClusters();
+ topology.getClusterAddrTable().remove("cluster-a");
+ topology.getBrokerAddrTable().remove("broker-a");
+ when(broker.examineBrokerClusterInfo()).thenReturn(topology);
+ TopicConfig physical = new TopicConfig("orders", 8, 8, 6);
+ when(broker.examineTopicConfig("10.0.1.1:10911", "orders")).thenReturn(physical);
+ TopicVO request = new TopicVO();
+ request.setName("orders");
+ request.setType(TopicType.NORMAL);
+ admin.importTopic(" first ", request);
+ assertThat(guard.check(first, topic, 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);
+ assertThat(physical.getReadQueueNums()).isEqualTo(8);
+ verify(broker, never()).createAndUpdateTopicConfig(anyString(), any());
+ assertThat(jdbc.queryForObject("SELECT read_queue_nums FROM rmq_instance_topic", Integer.class)).isEqualTo(8);
+ }
+
+ @ParameterizedTest
+ @EnumSource(TopicType.class)
+ void importEveryTopicTypeUsesPhysicalReadAttributeInsteadOfWriteAttributeTest(TopicType type) throws Exception {
+ var runtime = mock(RuntimeAdminClientResolver.class);
+ var broker = mock(DefaultMQAdminExt.class);
+ var admin = apacheAdmin(runtime, broker);
+ when(resolver.findByIdentifier("first")).thenReturn(Optional.of(first));
+ ClusterInfo topology = twoClusters();
+ topology.getClusterAddrTable().remove("cluster-a");
+ topology.getBrokerAddrTable().remove("broker-a");
+ when(broker.examineBrokerClusterInfo()).thenReturn(topology);
+ TopicConfig physical = new TopicConfig("orders", 8, 8, 6);
+ physical.setAttributes(Map.of("message.type", type.name()));
+ when(broker.examineTopicConfig("10.0.1.1:10911", "orders")).thenReturn(physical);
+ TopicVO request = new TopicVO();
+ request.setName("orders");
+ request.setType(type);
+ admin.importTopic("first", request);
+ assertThat(guard.check(first, topic, 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());
+ }
+
+ @Test
+ void importNewTopicCreatesOnlyAfterCommittedClaimAndRetainsPartialFailureTest() throws Exception {
+ var runtime = mock(RuntimeAdminClientResolver.class);
+ var broker = mock(DefaultMQAdminExt.class);
+ var admin = apacheAdmin(runtime, broker);
+ when(resolver.findByIdentifier("first")).thenReturn(Optional.of(first));
+ when(resolver.findByIdentifier("second")).thenReturn(Optional.of(second));
+ ClusterInfo topology = twoClusters();
+ topology.getClusterAddrTable().remove("cluster-a");
+ topology.getBrokerAddrTable().remove("broker-a");
+ when(broker.examineBrokerClusterInfo()).thenReturn(topology);
+ org.mockito.Mockito.doAnswer(call -> {
+ assertThat(guard.check(first, topic, true).instanceId()).isEqualTo("first");
+ throw new BusinessException(502, "Remote already partially succeeded");
+ }).when(broker).createAndUpdateTopicConfig(anyString(), any());
+ TopicVO request = new TopicVO();
+ request.setName("orders");
+ assertCode(() -> admin.importTopic("first", request), 502);
+ assertCode(() -> admin.importTopic("second", request), 409);
+ assertThat(jdbc.queryForObject("SELECT status FROM rmq_instance_topic", String.class)).isEqualTo("PENDING");
+ verify(broker).createAndUpdateTopicConfig(anyString(), any());
+ }
+
+ @Test
+ void importExistingGroupPreservesPhysicalSettingsAndRejectsDifferencesTest() throws Exception {
+ var runtime = mock(RuntimeAdminClientResolver.class);
+ var broker = mock(DefaultMQAdminExt.class);
+ var admin = apacheAdmin(runtime, broker);
+ when(resolver.findByIdentifier("first")).thenReturn(Optional.of(first));
+ ClusterInfo topology = twoClusters();
+ topology.getClusterAddrTable().remove("cluster-a");
+ topology.getBrokerAddrTable().remove("broker-a");
+ when(broker.examineBrokerClusterInfo()).thenReturn(topology);
+ SubscriptionGroupConfig physical = new SubscriptionGroupConfig();
+ physical.setGroupName("buyers");
+ physical.setRetryMaxTimes(16);
+ physical.setConsumeEnable(false);
+ physical.setRetryQueueNums(7);
+ when(broker.examineSubscriptionGroupConfig("10.0.1.1:10911", "buyers")).thenReturn(physical);
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("buyers");
+ group.setInstanceId("first");
+ admin.importConsumerGroup(group);
+ group.setRetryMaxTimes(20);
+ assertCode(() -> admin.importConsumerGroup(group), 409);
+ assertThat(physical.isConsumeEnable()).isFalse();
+ assertThat(physical.getRetryQueueNums()).isEqualTo(7);
+ verify(broker, never()).createAndUpdateSubscriptionGroupConfig(anyString(), any());
+ assertThat(jdbc.queryForObject("SELECT status FROM rmq_instance_group", String.class)).isEqualTo("ACTIVE");
+ }
+
+
+ @Test
+ void liteTtlUsesOwnedSecondClusterAndPreservesOwnershipOnFailureTest() throws Exception {
+ guard.write(first, topic, "cluster-b", true, "LITE", () -> null);
+ when(resolver.findByIdentifier("first")).thenReturn(Optional.of(first));
+ when(resolver.findByIdentifier("second")).thenReturn(Optional.of(second));
+ var runtime = mock(RuntimeAdminClientResolver.class);
+ var broker = mock(DefaultMQAdminExt.class);
+ var factory = mock(MqAdminExtFactory.class);
+ var lite = new RocketMQLiteTopicProvider(factory, new RocketMQProperties(), runtime, guard);
+ when(runtime.execute(any(InstanceVO.class), any())).thenAnswer(call ->
+ call.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(broker));
+ when(broker.examineBrokerClusterInfo()).thenReturn(twoClusters());
+ when(broker.examineTopicConfig("10.0.1.1:10911", "orders")).thenAnswer(call -> {
+ TopicConfig physical = new TopicConfig("orders");
+ physical.setAttributes(Map.of("message.type", "LITE"));
+ return physical;
+ });
+ lite.extendTTL(" first ", " orders ", 60_000L);
+ assertCode(() -> lite.extendTTL("second", "orders", 60_000L), 409);
+ verify(broker).createAndUpdateTopicConfig(eq("10.0.1.1:10911"), any());
+ verify(broker, never()).createAndUpdateTopicConfig(eq("10.0.0.1:10911"), any());
+ verify(broker, never()).createAndUpdateTopicConfig(eq("10.0.1.2:10911"), any());
+ org.mockito.Mockito.doThrow(new BusinessException(502, "TTL remote write failed"))
+ .when(broker).createAndUpdateTopicConfig(eq("10.0.1.1:10911"), any());
+ 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");
+ }
+
+ @Test
+ void virtualClaimAndRegistrationSerializeWithoutChangingOwnershipTest() throws Exception {
+ InstanceVO virtual = InstanceVO.builder().name("virtual").build();
+ 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", () -> {
+ 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));
+ 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();
+ }
+
+ @Test
+ void staleRegisteredSnapshotCannotWriteAfterConnectionChangeOrDeletionTest() {
+ AtomicInteger remote = new AtomicInteger();
+ InstanceVO updated = instance("first");
+ updated.setEndpoint("other:9876");
+ new TransactionTemplate(transactions).executeWithoutResult(status -> {
+ guard.lockForInstanceUpdate(first, updated);
+ jdbc.update("UPDATE rmq_instance SET endpoint = ? WHERE id = ?", updated.getEndpoint(), first.getId());
+ });
+ assertCode(() -> guard.write(first, topic, "cluster", true, "NORMAL", remote::incrementAndGet), 409);
+ new TransactionTemplate(transactions).executeWithoutResult(status -> {
+ guard.lockForInstanceDeletion(first.getId());
+ jdbc.update("DELETE FROM rmq_instance WHERE id = ?", first.getId());
+ });
+ assertCode(() -> guard.write(updated, topic, "cluster", true, "NORMAL", remote::incrementAndGet), 409);
+ assertThat(remote).hasValue(0);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_instance_topic", Integer.class)).isZero();
+ }
+
+ @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);
+ migration.afterPropertiesSet();
+ migration.afterPropertiesSet();
+ assertThat(jdbc.queryForObject("SELECT instance_id FROM rmq_instance_topic", String.class)).isEmpty();
+ assertCode(() -> guard.withInstanceRegistration("cluster", () -> null), 409);
+ }
+
+ @Test
+ void migrationConcurrentStartupIsIdempotentTest() throws Exception {
+ legacyIndexes();
+ jdbc.execute("DROP TABLE rmq_instance_ownership_lock");
+ CountDownLatch start = new CountDownLatch(1);
+ try (var executor = Executors.newFixedThreadPool(2)) {
+ java.util.concurrent.Callable<Void> migrate = () -> {
+ await(start);
+ new ResourceOwnershipSchemaMigration(dataSource, defaultClusters).afterPropertiesSet();
+ return null;
+ };
+ var a = executor.submit(migrate);
+ var b = executor.submit(migrate);
+ start.countDown();
+ a.get(10, TimeUnit.SECONDS);
+ b.get(10, TimeUnit.SECONDS);
+ }
+ guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
+ assertCode(() -> guard.write(second, topic, "cluster", true, "NORMAL", () -> null), 409);
+ }
+
+ @Test
+ void legacyApacheVendorSnapshotAcceptsBlankButRejectsChangedVendorTest() {
+ for (String vendor : new String[] {null, "", " "}) {
+ jdbc.update("UPDATE rmq_instance SET vendor = ? WHERE name = 'first'", vendor);
+ guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
+ }
+ jdbc.update("UPDATE rmq_instance SET vendor = 'ALIYUN' WHERE name = 'first'");
+ AtomicInteger remote = new AtomicInteger();
+ assertCode(() -> guard.write(first, topic, "cluster", true, "NORMAL", remote::incrementAndGet), 409);
+ assertThat(remote).hasValue(0);
+ }
+
+ @Test
+ void registrationWinningRacePreventsStaleVirtualWriteTest() {
+ guard.withInstanceRegistration("virtual", () -> jdbc.update(
+ "INSERT INTO rmq_instance (name, type, endpoint) VALUES ('virtual','DIRECT','other:9876')"));
+ AtomicInteger remote = new AtomicInteger();
+ assertCode(() -> guard.write(InstanceVO.builder().name("virtual").build(), topic,
+ "virtual", true, "NORMAL", remote::incrementAndGet), 409);
+ assertThat(remote).hasValue(0);
+ }
+
+ private RocketMQAdminClientImpl apacheAdmin(RuntimeAdminClientResolver runtime, DefaultMQAdminExt broker) {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), RmqTopic.class);
+ when(runtime.execute(anyString(), any())).thenAnswer(call ->
+ call.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(broker));
+ RmqTopicMapper mapper = mock(RmqTopicMapper.class);
+ when(mapper.selectOne(any())).thenAnswer(call -> jdbc.query(
+ "SELECT id, name, instance_id, cluster_id, topic_type FROM rmq_instance_topic WHERE name = 'orders'",
+ (rs, row) -> {
+ RmqTopic entity = new RmqTopic();
+ entity.setId(rs.getLong(1));
+ entity.setName(rs.getString(2));
+ entity.setInstanceId(rs.getString(3));
+ entity.setClusterId(rs.getString(4));
+ entity.setTopicType(rs.getString(5));
+ return entity;
+ }).stream().findFirst().orElse(null));
+ when(mapper.updateById(any(RmqTopic.class))).thenAnswer(call -> {
+ RmqTopic entity = call.getArgument(0);
+ return jdbc.update("UPDATE rmq_instance_topic SET status = ?, read_queue_nums = ?, write_queue_nums = ? WHERE id = ?",
+ entity.getStatus(), entity.getReadQueueNums(), entity.getWriteQueueNums(), entity.getId());
+ });
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), RmqGroup.class);
+ RmqGroupMapper groups = mock(RmqGroupMapper.class);
+ when(groups.selectOne(any())).thenAnswer(call -> jdbc.query(
+ "SELECT id, name, instance_id, cluster_id FROM rmq_instance_group WHERE name = 'buyers'", (rs, row) -> {
+ RmqGroup entity = new RmqGroup();
+ entity.setId(rs.getLong(1));
+ entity.setName(rs.getString(2));
+ entity.setInstanceId(rs.getString(3));
+ entity.setClusterId(rs.getString(4));
+ return entity;
+ }).stream().findFirst().orElse(null));
+ when(groups.updateById(any(RmqGroup.class))).thenAnswer(call -> {
+ RmqGroup entity = call.getArgument(0);
+ return jdbc.update("UPDATE rmq_instance_group SET status = ?, max_retry = ? WHERE id = ?",
+ entity.getStatus(), entity.getMaxRetry(), entity.getId());
+ });
+ return new RocketMQAdminClientImpl(mock(MqAdminExtFactory.class), new RocketMQProperties(), mapper,
+ groups, mock(AuditService.class), runtime, mock(MqClientPool.class), guard);
+ }
+
+ private ClusterInfo twoClusters() {
+ ClusterInfo info = new ClusterInfo();
+ info.setClusterAddrTable(new HashMap<>(Map.of("cluster-a", Set.of("broker-a"), "cluster-b", Set.of("broker-b"))));
+ info.setBrokerAddrTable(new HashMap<>(Map.of(
+ "broker-a", new BrokerData("cluster-a", "broker-a", new HashMap<>(Map.of(0L, "10.0.0.1:10911"))),
+ "broker-b", new BrokerData("cluster-b", "broker-b",
+ new HashMap<>(Map.of(0L, "10.0.1.1:10911", 1L, "10.0.1.2:10911"))))));
+ return info;
+ }
+
+ private void legacyIndexes() {
+ for (String kind : List.of("topic", "group")) {
+ jdbc.execute("ALTER TABLE rmq_instance_" + kind + " DROP INDEX uk_" + kind + "_name");
+ jdbc.execute("CREATE UNIQUE INDEX uk_cluster_instance_" + kind + " ON rmq_instance_" + kind
+ + " (cluster_id, instance_id, name)");
+ }
+ }
+
+ private int claim(CountDownLatch start, InstanceVO instance, AtomicInteger remote) {
+ await(start);
+ try {
+ guard.write(instance, topic, "cluster", true, "NORMAL", remote::incrementAndGet);
+ return 200;
+ } catch (BusinessException failure) {
+ return failure.getCode();
+ }
+ }
+
+ private InstanceVO instance(String name) {
+ InstanceVO instance = InstanceVO.builder().name(name).vendor(InstanceVendor.APACHE)
+ .type(InstanceType.DIRECT).endpoint("same:9876").build();
+ instance.setId(jdbc.queryForObject("SELECT id FROM rmq_instance WHERE name = ?", Long.class, name));
+ return instance;
+ }
+
+ private static void assertCode(Runnable action, int code) {
+ assertThatThrownBy(action::run).isInstanceOf(BusinessException.class)
+ .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(code));
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ if (!latch.await(5, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting for concurrency test");
+ }
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(interrupted);
+ }
+ }
+}
diff --git a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
index 4609224bc..1e5226600 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
@@ -41,7 +41,6 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
@@ -413,19 +412,36 @@ class ConsumerGroupControllerTest extends WebMvcAuthTestSupport {
void deleteConsumerGroupShouldReturnSuccess() throws Exception {
mockMvc.perform(post("/api/groups/delete")
.contentType(MediaType.APPLICATION_JSON)
- .content(objectMapper.writeValueAsString(Map.of("name", "cg-orders"))))
+ .content(objectMapper.writeValueAsString(Map.of("instanceId", "instance-a", "name", "cg-orders"))))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.message").value("success"));
- verify(metadataService).deleteConsumerGroup(isNull(), eq("cg-orders"));
+ verify(metadataService).deleteConsumerGroup(eq("instance-a"), eq("cg-orders"));
+ }
+
+ @Test
+ void deleteGroupRequiresInstanceTest() throws Exception {
+ mockMvc.perform(post("/api/groups/delete").contentType(MediaType.APPLICATION_JSON)
+ .content("{\"name\":\"orders\"}"))
+ .andExpect(status().isBadRequest()).andExpect(jsonPath("$.code").value(400));
+ verifyNoInteractions(metadataService);
+ }
+
+ @Test
+ void deleteGroupReportsOwnershipConflictTest() throws Exception {
+ org.mockito.Mockito.doThrow(new org.apache.rocketmq.studio.common.exception.BusinessException(409, "Ownership conflict"))
+ .when(metadataService).deleteConsumerGroup("instance-a", "orders");
+ mockMvc.perform(post("/api/groups/delete").contentType(MediaType.APPLICATION_JSON)
+ .content("{\"instanceId\":\"instance-a\",\"name\":\"orders\"}"))
+ .andExpect(status().isConflict()).andExpect(jsonPath("$.code").value(409));
}
@Test
void deleteConsumerGroupShouldRejectMissingName() throws Exception {
mockMvc.perform(post("/api/groups/delete")
.contentType(MediaType.APPLICATION_JSON)
- .content("{}"))
+ .content("{\"instanceId\":\"instance-a\"}"))
.andExpect(status().isBadRequest())
.andExpect(jsonPath("$.code").value(400))
.andExpect(jsonPath("$.message").value("name is required"));
@@ -437,7 +453,7 @@ class ConsumerGroupControllerTest extends WebMvcAuthTestSupport {
void deleteConsumerGroupShouldRejectBlankName() throws Exception {
mockMvc.perform(post("/api/groups/delete")
.contentType(MediaType.APPLICATION_JSON)
- .content(objectMapper.writeValueAsString(Map.of("name", " "))))
+ .content(objectMapper.writeValueAsString(Map.of("instanceId", "instance-a", "name", " "))))
.andExpect(status().isBadRequest())
.andExpect(jsonPath("$.code").value(400))
.andExpect(jsonPath("$.message").value("name is required"));
diff --git a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
index 58d5267e9..260123acd 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
@@ -7,10 +7,19 @@
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
*/
package org.apache.rocketmq.studio.instance.message;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
+import static org.mockito.ArgumentMatchers.any;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.audit.OperationAuditService;
@@ -28,11 +37,33 @@ import static org.mockito.Mockito.when;
class MessageServiceTest {
+ private ResourceOwnershipGuard ownershipGuard() {
+ ResourceOwnershipGuard guard = mock(ResourceOwnershipGuard.class);
+ when(guard.requireInstance(any())).thenAnswer(call -> InstanceVO.builder()
+ .name(ResourceOwnershipGuard.requireText(call.getArgument(0), "instanceId")).build());
+ when(guard.topicResource(any())).thenCallRealMethod();
+ when(guard.withOwned(any(), any(), any())).thenAnswer(call ->
+ call.<java.util.function.Supplier<Object>>getArgument(2).get());
+ return guard;
+ }
+
+ @Test
+ void directConsumeRejectsMissingInstanceBeforeProviderTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class),
+ mock(OperationAuditService.class), ownershipGuard());
+ assertThatThrownBy(() -> service.consumeMessageDirectly(new DirectConsumeMessageDTO()))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(400));
+ verifyNoInteractions(provider, registry);
+ }
+
@Test
void rejectsNegativeQueueCoordinatesBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
MessageService service = new MessageService(provider, mock(InstanceProviderRegistry.class),
- mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
assertThatThrownBy(() -> service.pullMessageAtOffset("instance-a", "TopicA", "broker-a", -1, 0))
.isInstanceOf(BusinessException.class)
@@ -48,7 +79,7 @@ class MessageServiceTest {
void rejectsKeyQueryWithoutTopicBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
- MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
assertThatThrownBy(() -> service.queryMessages(null, null, null, null, "order-1", null, null))
.isInstanceOf(BusinessException.class)
@@ -61,7 +92,7 @@ class MessageServiceTest {
void rejectsMessageIdQueryWithoutTopicBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
- MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
assertThatThrownBy(() -> service.queryMessages(null, null, "msg-001", null, null, null, null))
.isInstanceOf(BusinessException.class)
@@ -74,7 +105,7 @@ class MessageServiceTest {
void rejectsBlankMessageTraceIdBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
- MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
assertThatThrownBy(() -> service.getMessageTrace("instance-a", " ", null))
.isInstanceOf(BusinessException.class)
@@ -87,7 +118,7 @@ class MessageServiceTest {
void rejectsReversedTopicQueryWindowBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
- MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
assertThatThrownBy(() -> service.queryMessages("instance-a", "TopicA", null, null, null, 200L, 100L))
.isInstanceOf(BusinessException.class)
@@ -100,7 +131,7 @@ class MessageServiceTest {
void rejectsTopicQueryWindowLongerThanSevenDaysBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
- MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
assertThatThrownBy(() -> service.queryMessages("instance-a", "TopicA", null, null, null, 0L,
8L * 24 * 60 * 60 * 1000))
@@ -116,7 +147,7 @@ class MessageServiceTest {
InstanceProvider provider = mock(InstanceProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
QueryHistoryService history = mock(QueryHistoryService.class);
- MessageService service = new MessageService(fallback, registry, history, mock(OperationAuditService.class));
+ MessageService service = new MessageService(fallback, registry, history, mock(OperationAuditService.class), ownershipGuard());
when(registry.byInstanceId("cloud-instance")).thenReturn(Optional.of(provider));
when(provider.queryMessagesDetailed("cloud-instance", "orders", null, null, "ORDER-1", null, null))
.thenReturn(MessageQueryResult.complete(
@@ -145,7 +176,7 @@ class MessageServiceTest {
.consumeResult("CR_SUCCESS").build();
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
when(provider.consumeMessageDirectly(request)).thenReturn(expected);
- MessageService service = new MessageService(fallback, registry, mock(QueryHistoryService.class), audit);
+ MessageService service = new MessageService(fallback, registry, mock(QueryHistoryService.class), audit, ownershipGuard());
org.assertj.core.api.Assertions.assertThat(service.consumeMessageDirectly(request)).isSameAs(expected);
@@ -170,7 +201,7 @@ class MessageServiceTest {
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
when(provider.consumeMessageDirectly(request)).thenReturn(DirectConsumeMessageResultVO.builder()
.consumeResult("CR_ROLLBACK").build());
- MessageService service = new MessageService(fallback, registry, mock(QueryHistoryService.class), audit);
+ MessageService service = new MessageService(fallback, registry, mock(QueryHistoryService.class), audit, ownershipGuard());
service.consumeMessageDirectly(request);
@@ -194,7 +225,7 @@ class MessageServiceTest {
request.setClientId("client-a");
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
when(provider.consumeMessageDirectly(request)).thenThrow(new IllegalStateException("client offline"));
- MessageService service = new MessageService(fallback, registry, mock(QueryHistoryService.class), audit);
+ MessageService service = new MessageService(fallback, registry, mock(QueryHistoryService.class), audit, ownershipGuard());
assertThatThrownBy(() -> service.consumeMessageDirectly(request))
.isInstanceOf(IllegalStateException.class)
@@ -211,7 +242,7 @@ class MessageServiceTest {
void rejectsOverflowingTopicQueryWindowBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
- MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
assertThatThrownBy(() -> service.queryMessages("instance-a", "TopicA", null, null, null,
0L, Long.MAX_VALUE))
@@ -226,7 +257,7 @@ class MessageServiceTest {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
QueryHistoryService history = mock(QueryHistoryService.class);
- MessageService service = new MessageService(provider, registry, history, mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, history, mock(OperationAuditService.class), ownershipGuard());
when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
when(provider.queryMessagesDetailed("instance-a", "TopicA", null, null, null, 1000L, 2000L))
.thenReturn(MessageQueryResult.complete(java.util.stream.IntStream.range(0, 200)
@@ -246,7 +277,7 @@ class MessageServiceTest {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
QueryHistoryService history = mock(QueryHistoryService.class);
- MessageService service = new MessageService(provider, registry, history, mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, history, mock(OperationAuditService.class), ownershipGuard());
when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
when(provider.queryMessagesDetailed("instance-a", "TopicA", null, null, null, 1000L, 2000L))
.thenReturn(MessageQueryResult.truncated(
@@ -266,7 +297,7 @@ class MessageServiceTest {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
QueryHistoryService history = mock(QueryHistoryService.class);
- MessageService service = new MessageService(provider, registry, history, mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, history, mock(OperationAuditService.class), ownershipGuard());
when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
when(provider.queryMessagesDetailed("instance-a", "TopicA", null, null, "order-1", 1000L, 2000L))
.thenReturn(MessageQueryResult.truncated(List.of(
@@ -283,7 +314,7 @@ class MessageServiceTest {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
QueryHistoryService history = mock(QueryHistoryService.class);
- MessageService service = new MessageService(provider, registry, history, mock(OperationAuditService.class));
+ MessageService service = new MessageService(provider, registry, history, mock(OperationAuditService.class), ownershipGuard());
when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
when(provider.queryMessagesDetailed("instance-a", "TopicA", null, null, null, 1000L, 2000L))
.thenReturn(MessageQueryResult.complete(
@@ -305,7 +336,7 @@ class MessageServiceTest {
MessageProvider fallback = mock(MessageProvider.class);
InstanceProvider provider = mock(InstanceProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
- MessageService service = new MessageService(fallback, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ MessageService service = new MessageService(fallback, registry, mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
TraceRecordVO trace = TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
when(provider.getMessageTrace("instance-a", "msg-001", "orders")).thenReturn(trace);
@@ -325,7 +356,7 @@ class MessageServiceTest {
InstanceProvider provider = mock(InstanceProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
QueryHistoryService history = mock(QueryHistoryService.class);
- MessageService service = new MessageService(fallback, registry, history, mock(OperationAuditService.class));
+ MessageService service = new MessageService(fallback, registry, history, mock(OperationAuditService.class), ownershipGuard());
TraceRecordVO trace = TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
when(provider.getMessageTrace("instance-a", "msg-001", "orders", "CUSTOM_TRACE"))
@@ -344,7 +375,7 @@ class MessageServiceTest {
MessageProvider fallback = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
QueryHistoryService history = mock(QueryHistoryService.class);
- MessageService service = new MessageService(fallback, registry, history, mock(OperationAuditService.class));
+ MessageService service = new MessageService(fallback, registry, history, mock(OperationAuditService.class), ownershipGuard());
TraceRecordVO trace = TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
when(fallback.getMessageTraceByKey("instance-a", "ORDER-1", "orders", null)).thenReturn(trace);
@@ -363,7 +394,7 @@ class MessageServiceTest {
MessageProvider fallback = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
QueryHistoryService history = mock(QueryHistoryService.class);
- MessageService service = new MessageService(fallback, registry, history, mock(OperationAuditService.class));
+ MessageService service = new MessageService(fallback, registry, history, mock(OperationAuditService.class), ownershipGuard());
TraceRecordVO trace = TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
when(fallback.getMessageTraceByKey("instance-a", "ORDER-1", "orders", "CUSTOM_TRACE"))
@@ -380,7 +411,7 @@ class MessageServiceTest {
void rejectsBlankUniqueKeyQueryBeforeCallingProviderTest() {
MessageProvider provider = mock(MessageProvider.class);
MessageService service = new MessageService(provider, mock(InstanceProviderRegistry.class),
- mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
assertThatThrownBy(() -> service.queryMessageByUniqueKey("instance-a", null, "uniq-1", null, null))
.isInstanceOf(BusinessException.class)
@@ -396,7 +427,7 @@ class MessageServiceTest {
void delegatesUniqueKeyQueryToMessageProviderTest() {
MessageProvider provider = mock(MessageProvider.class);
MessageService service = new MessageService(provider, mock(InstanceProviderRegistry.class),
- mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ mock(QueryHistoryService.class), mock(OperationAuditService.class), ownershipGuard());
MessageRecordVO record = MessageRecordVO.builder().msgId("msg-1").build();
when(provider.queryMessageByUniqueKey("instance-a", "TopicA", "uniq-1", 100L, 200L))
.thenReturn(List.of(record));
diff --git a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/LiteTopicControllerTest.java b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/LiteTopicControllerTest.java
index defe06f24..e8e8de4e0 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/LiteTopicControllerTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/LiteTopicControllerTest.java
@@ -119,9 +119,27 @@ class LiteTopicControllerTest extends WebMvcAuthTestSupport {
.andExpect(jsonPath("$.data.popProgress").value(96));
}
+ @Test
+ void ttlRequiresExplicitInstanceTest() throws Exception {
+ mockMvc.perform(post("/api/liteTopic/extendTTL").contentType(MediaType.APPLICATION_JSON)
+ .content("{\"topicPattern\":\"chat\",\"newTTL\":60000}"))
+ .andExpect(status().isBadRequest()).andExpect(jsonPath("$.code").value(400));
+ verifyNoInteractions(liteTopicService);
+ }
+
+ @Test
+ void ttlReportsOwnershipConflictTest() throws Exception {
+ doThrow(new BusinessException(409, "Ownership conflict")).when(liteTopicService)
+ .extendTTL("instance-a", "chat", 60_000L);
+ mockMvc.perform(post("/api/liteTopic/extendTTL").contentType(MediaType.APPLICATION_JSON)
+ .content("{\"instanceId\":\"instance-a\",\"topicPattern\":\"chat\",\"newTTL\":60000}"))
+ .andExpect(status().isConflict()).andExpect(jsonPath("$.code").value(409));
+ }
+
@Test
void extendTTLShouldDelegateToService() throws Exception {
LiteTopicTTLUpdateDTO request = new LiteTopicTTLUpdateDTO();
+ request.setInstanceId("instance-a");
request.setTopicPattern("chat/{sessionId}");
request.setNewTTL(7_200_000L);
@@ -131,17 +149,18 @@ class LiteTopicControllerTest extends WebMvcAuthTestSupport {
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200));
- verify(liteTopicService).extendTTL(eq("chat/{sessionId}"), eq(7_200_000L));
+ verify(liteTopicService).extendTTL(eq("instance-a"), eq("chat/{sessionId}"), eq(7_200_000L));
}
@Test
void extendTTLShouldReturnUnsupportedWhenProviderIsUnavailable() throws Exception {
LiteTopicTTLUpdateDTO request = new LiteTopicTTLUpdateDTO();
+ request.setInstanceId("instance-a");
request.setTopicPattern("chat/{sessionId}");
request.setNewTTL(7_200_000L);
doThrow(new BusinessException(501, "LiteTopic provider integration is not available"))
.when(liteTopicService)
- .extendTTL(eq("chat/{sessionId}"), eq(7_200_000L));
+ .extendTTL(eq("instance-a"), eq("chat/{sessionId}"), eq(7_200_000L));
mockMvc.perform(post("/api/liteTopic/extendTTL")
.contentType(MediaType.APPLICATION_JSON)
@@ -154,6 +173,7 @@ class LiteTopicControllerTest extends WebMvcAuthTestSupport {
@Test
void extendTTLShouldRejectMissingTopicPattern() throws Exception {
LiteTopicTTLUpdateDTO request = new LiteTopicTTLUpdateDTO();
+ request.setInstanceId("instance-a");
request.setNewTTL(7_200_000L);
mockMvc.perform(post("/api/liteTopic/extendTTL")
@@ -169,6 +189,7 @@ class LiteTopicControllerTest extends WebMvcAuthTestSupport {
@Test
void extendTTLShouldRejectNonPositiveTTL() throws Exception {
LiteTopicTTLUpdateDTO request = new LiteTopicTTLUpdateDTO();
+ request.setInstanceId("instance-a");
request.setTopicPattern("chat/{sessionId}");
request.setNewTTL(0L);
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 f5262dff2..d4d2b51c1 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
@@ -18,6 +18,9 @@
package org.apache.rocketmq.studio.instance.topic;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.model.LiteTopicQuota;
import org.apache.rocketmq.studio.model.LiteTopicSession;
@@ -39,8 +42,10 @@ import static org.mockito.Mockito.when;
class LiteTopicServiceTest {
/** No NameServer configured: the provider reports the feature as unavailable. */
+ private final ResourceOwnershipGuard guard = mock(ResourceOwnershipGuard.class);
private final LiteTopicService unavailableService = new LiteTopicService(
- new RocketMQLiteTopicProvider(new MqAdminExtFactory(), new RocketMQProperties()));
+ new RocketMQLiteTopicProvider(new MqAdminExtFactory(), new RocketMQProperties(),
+ mock(RuntimeAdminClientResolver.class), guard), guard);
@Test
void listLiteTopicsShouldReportUnsupportedWhenProviderIsUnavailable() {
@@ -71,23 +76,30 @@ class LiteTopicServiceTest {
@Test
void extendTTLShouldRejectInvalidInput() {
- assertThatThrownBy(() -> unavailableService.extendTTL("", 1L))
+ assertThatThrownBy(() -> unavailableService.extendTTL("instance-a", "", 1L))
.isInstanceOf(BusinessException.class)
.hasMessage("topicPattern is required")
.satisfies(ex -> assertThat(((BusinessException) ex).getCode()).isEqualTo(400));
- assertThatThrownBy(() -> unavailableService.extendTTL("chat/{sessionId}", 0L))
+ assertThatThrownBy(() -> unavailableService.extendTTL("instance-a", "chat/{sessionId}", 0L))
.isInstanceOf(BusinessException.class)
.hasMessage("newTTL must be positive")
.satisfies(ex -> assertThat(((BusinessException) ex).getCode()).isEqualTo(400));
}
@Test
- void extendTTLShouldReportUnsupportedWhenProviderIsUnavailable() {
- assertThatThrownBy(() -> unavailableService.extendTTL("chat/{sessionId}", 7_200_000L))
- .isInstanceOfSatisfying(BusinessException.class, ex -> {
- assertThat(ex.getCode()).isEqualTo(501);
- assertThat(ex.getMessage()).isEqualTo(LiteTopicProvider.UNSUPPORTED);
- });
+ void extendTTLRequiresInstanceAndOwnershipTest() {
+ LiteTopicProvider provider = mock(LiteTopicProvider.class);
+ LiteTopicService service = new LiteTopicService(provider, guard);
+ assertThatThrownBy(() -> service.extendTTL(null, "chat", 60_000L))
+ .isInstanceOfSatisfying(BusinessException.class, ex -> assertThat(ex.getCode()).isEqualTo(400));
+ InstanceVO instance = InstanceVO.builder().name("instance-a").build();
+ 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"));
+ assertThatThrownBy(() -> service.extendTTL(" instance-a ", " chat ", 60_000L))
+ .isInstanceOfSatisfying(BusinessException.class, ex -> assertThat(ex.getCode()).isEqualTo(409));
+ org.mockito.Mockito.verifyNoInteractions(provider);
}
@Test
@@ -117,7 +129,7 @@ class LiteTopicServiceTest {
when(provider.listLiteTopics("chat", "DEFAULT")).thenReturn(List.of(summary));
when(provider.isSupported()).thenReturn(true);
- LiteTopicService service = new LiteTopicService(provider);
+ LiteTopicService service = new LiteTopicService(provider, guard);
List<LiteTopicItemVO> items = service.listLiteTopics("chat", "DEFAULT");
assertThat(items).singleElement().satisfies(item -> {
@@ -154,7 +166,7 @@ class LiteTopicServiceTest {
session.setLiteTopics(new LinkedHashSet<>(List.of("bob", "alice")));
when(provider.getSession("chat~g~c1")).thenReturn(session);
- LiteTopicSessionVO vo = new LiteTopicService(provider).getSession("chat~g~c1");
+ LiteTopicSessionVO vo = new LiteTopicService(provider, guard).getSession("chat~g~c1");
assertThat(vo.getSessionId()).isEqualTo("chat~g~c1");
assertThat(vo.getClientAddress()).isEqualTo("10.0.0.9:1234");
@@ -187,7 +199,7 @@ class LiteTopicServiceTest {
quota.setMaxCreationRate(0.0);
when(provider.getQuota("DEFAULT")).thenReturn(quota);
- LiteTopicQuotaVO vo = new LiteTopicService(provider).getQuota("DEFAULT");
+ LiteTopicQuotaVO vo = new LiteTopicService(provider, guard).getQuota("DEFAULT");
assertThat(vo.getCurrentTopicCount()).isEqualTo(10);
assertThat(vo.getMaxTopicCount()).isEqualTo(40);
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 e6148c9bc..e2535c047 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
@@ -25,6 +25,8 @@ import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.apache.rocketmq.studio.audit.OperationAuditService;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.domain.PageResult;
@@ -63,15 +65,14 @@ import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -106,6 +107,9 @@ class MetadataServiceTest {
@Mock
private RuntimeAdminClientResolver runtimeAdminClientResolver;
+ @Mock
+ private ResourceOwnershipGuard ownershipGuard;
+
@InjectMocks
private MetadataService metadataService;
@@ -224,7 +228,16 @@ class MetadataServiceTest {
}
@BeforeEach
- void routeBlankInstanceIdsToApacheProvider() {
+ void configureReadRoutingAndExplicitWriteOwnership() {
+ lenient().when(ownershipGuard.requireInstance(any())).thenAnswer(call -> {
+ String name = ResourceOwnershipGuard.requireText(call.getArgument(0), "instanceId");
+ return InstanceVO.builder().name(name).vendor("cloud-instance".equals(name)
+ ? InstanceVendor.TENCENT : InstanceVendor.APACHE).build();
+ });
+ lenient().when(ownershipGuard.topicResource(any())).thenCallRealMethod();
+ lenient().doCallRealMethod().when(ownershipGuard).requireSupportedProvider(any());
+ lenient().when(ownershipGuard.withOwned(any(), any(), any())).thenAnswer(call ->
+ call.<java.util.function.Supplier<Object>>getArgument(2).get());
lenient().when(providerRegistry.forVendor(InstanceVendor.APACHE)).thenReturn(apacheProvider);
lenient().when(apacheProvider.vendor()).thenReturn(InstanceVendor.APACHE);
lenient().when(cloudProvider.vendor()).thenReturn(InstanceVendor.TENCENT);
@@ -353,8 +366,8 @@ class MetadataServiceTest {
verify(apacheProvider).createConsumerGroup("instance-a", group);
verify(apacheProvider).updateConsumerGroup("instance-a", group);
verify(apacheProvider).deleteConsumerGroup("instance-a", "consumers");
- // decision 17: group deletion cascades to the dead-letter topic
- verify(apacheProvider).deleteTopic("instance-a", "%DLQ%consumers");
+ // Derived cleanup must happen within the underlying group ownership lock; the common layer does not re-issue the delete.
+ verify(apacheProvider, never()).deleteTopic("instance-a", "%DLQ%consumers");
verifyNoInteractions(operationAuditService);
}
@@ -362,6 +375,7 @@ class MetadataServiceTest {
void createTopicShouldDelegateToApacheProvider() {
TopicVO input = new TopicVO();
input.setName("new-topic");
+ input.setInstanceId("instance-a");
input.setWriteQueues(8);
input.setReadQueues(8);
@@ -375,7 +389,7 @@ class MetadataServiceTest {
TopicVO result = metadataService.createTopic(input);
assertThat(result.getName()).isEqualTo("new-topic");
- verify(apacheProvider).createTopic(null, input);
+ verify(apacheProvider).createTopic("instance-a", input);
verifyNoInteractions(operationAuditService);
}
@@ -424,67 +438,6 @@ class MetadataServiceTest {
verifyNoInteractions(operationAuditService);
}
- @ParameterizedTest
- @ValueSource(booleans = {false, true})
- void cloudTopicWriteOperationsShouldRecordServiceBoundaryAudit(boolean explicitInstance) {
- TopicVO topic = new TopicVO();
- topic.setName("orders");
- topic.setInstanceId("cloud-instance");
- topic.setWriteQueues(4);
- topic.setReadQueues(4);
- if (explicitInstance) {
- metadataService.createTopic("cloud-instance", topic);
- metadataService.updateTopic("cloud-instance", topic);
- metadataService.deleteTopic("cloud-instance", " orders ");
- } else {
- metadataService.createTopic(topic);
- metadataService.updateTopic(topic);
- metadataService.deleteTopic("cloud-instance", " orders ");
- }
-
- verify(operationAuditService).record("CREATE_TOPIC", "TOPIC", "orders", "cloud-instance",
- "type=-, writeQueues=4, readQueues=4, perm=-", "SUCCESS", null);
- verify(operationAuditService).record("UPDATE_TOPIC", "TOPIC", "orders", "cloud-instance",
- "type=-, writeQueues=4, readQueues=4, perm=-", "SUCCESS", null);
- verify(operationAuditService).record("DELETE_TOPIC", "TOPIC", "orders", "cloud-instance",
- null, "SUCCESS", null);
- verifyNoMoreInteractions(operationAuditService);
- }
-
- @Test
- void auditFailureShouldNotAbortCloudMetadataOperation() {
- doThrow(new RuntimeException("audit unavailable")).when(operationAuditService)
- .record("DELETE_TOPIC", "TOPIC", "orders", "cloud-instance", null, "SUCCESS", null);
-
- metadataService.deleteTopic("cloud-instance", "orders");
-
- verify(cloudProvider).deleteTopic("cloud-instance", "orders");
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {false, true})
- void failedCloudMetadataOperationShouldRecordFailedAuditTest(boolean explicitInstance) {
- TopicVO topic = new TopicVO();
- topic.setName("orders");
- topic.setInstanceId("cloud-instance");
- topic.setWriteQueues(4);
- topic.setReadQueues(4);
- when(cloudProvider.createTopic("cloud-instance", topic))
- .thenThrow(new BusinessException(502, "open api unavailable"));
-
- assertThatThrownBy(() -> {
- if (explicitInstance) {
- metadataService.createTopic("cloud-instance", topic);
- } else {
- metadataService.createTopic(topic);
- }
- })
- .isInstanceOf(BusinessException.class);
-
- verify(operationAuditService).record("CREATE_TOPIC", "TOPIC", "orders", "cloud-instance",
- "type=-, writeQueues=4, readQueues=4, perm=-", "FAILED", "open api unavailable");
- verifyNoMoreInteractions(operationAuditService);
- }
@Test
void exportTopicsShouldApplyFiltersSelectedNamesSortingAndCsvEscaping() {
@@ -505,8 +458,8 @@ class MetadataServiceTest {
}
@Test
- void importTopicsShouldContinueAfterRowFailure() {
- when(apacheProvider.createTopic(eq("instance-a"), any(TopicVO.class))).thenAnswer(invocation -> {
+ void importTopicsShouldContinueAfterRowFailureTest() {
+ when(apacheProvider.importTopic(eq("instance-a"), any(TopicVO.class))).thenAnswer(invocation -> {
TopicVO topic = invocation.getArgument(1);
if ("topic-fail".equals(topic.getName())) {
throw new BusinessException(500, "broker rejected topic");
@@ -527,16 +480,71 @@ class MetadataServiceTest {
assertThat(result.getFailures().get(0).getMessage()).isEqualTo("broker rejected topic");
ArgumentCaptor<TopicVO> captor = ArgumentCaptor.forClass(TopicVO.class);
- verify(apacheProvider, org.mockito.Mockito.times(2)).createTopic(eq("instance-a"), captor.capture());
+ verify(apacheProvider, org.mockito.Mockito.times(2)).importTopic(eq("instance-a"), captor.capture());
+ verify(apacheProvider, never()).createTopic(anyString(), any());
assertThat(captor.getAllValues()).extracting(TopicVO::getInstanceId)
.containsExactly("instance-a", "instance-a");
}
@Test
- void deleteTopicShouldDelegateToApacheProvider() {
- metadataService.deleteTopic("topic-to-delete");
+ void missingInstanceRejectsAllPublicWritesBeforeProviderTest() {
+ TopicVO topic = topic("orders", null, TopicType.NORMAL);
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("buyers");
+ assertCode(() -> metadataService.createTopic(topic), 400);
+ assertCode(() -> metadataService.updateTopic(topic), 400);
+ assertCode(() -> metadataService.deleteTopic("orders"), 400);
+ assertCode(() -> metadataService.createConsumerGroup(group), 400);
+ assertCode(() -> metadataService.updateConsumerGroup(group), 400);
+ assertCode(() -> metadataService.deleteConsumerGroup("buyers"), 400);
+ assertCode(() -> metadataService.resetOffset("buyers", 1L, "orders"), 400);
+ assertCode(() -> metadataService.skipAccumulated(null, "buyers", "orders"), 400);
+ assertCode(() -> metadataService.importTopics(null, List.of(topicImportRequest("orders", null))), 400);
+ assertCode(() -> metadataService.importConsumerGroups(" ", List.of(importRequest("buyers", null))), 400);
+ assertCode(() -> metadataService.sendMessage(SendMessageDTO.builder().topic("orders").body("x").build()), 400);
+ verifyNoInteractions(apacheProvider, cloudProvider, adminClient, runtimeAdminClientResolver);
+ }
- verify(apacheProvider).deleteTopic(null, "topic-to-delete");
+ @Test
+ void crossInstanceWritesStopBeforeProviderTest() {
+ when(ownershipGuard.check(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);
+ 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);
+ }
+
+ private static void assertCode(Runnable operation, int code) {
+ assertThatThrownBy(operation::run).isInstanceOfSatisfying(BusinessException.class,
+ error -> assertThat(error.getCode()).isEqualTo(code));
}
@Test
@@ -628,16 +636,11 @@ class MetadataServiceTest {
}
@Test
- void deleteConsumerGroupShouldNotFailWhenDlqDeletionFailsTest() {
- // Decision 17: DLQ cascade is best-effort โ a missing/undeletable DLQ never blocks group deletion.
- doThrow(new BusinessException(502, "no route for %DLQ%group-a"))
- .when(apacheProvider).deleteTopic("instance-a", "%DLQ%group-a");
-
- assertThatCode(() -> metadataService.deleteConsumerGroup("instance-a", "group-a"))
- .doesNotThrowAnyException();
-
- verify(apacheProvider).deleteConsumerGroup("instance-a", "group-a");
- verify(apacheProvider).deleteTopic("instance-a", "%DLQ%group-a");
+ void deleteConsumerGroupPropagatesPartialCleanupFailureTest() {
+ doThrow(new BusinessException(502, "Derived resource cleanup failed; reservation kept"))
+ .when(apacheProvider).deleteConsumerGroup("instance-a", "group-a");
+ assertCode(() -> metadataService.deleteConsumerGroup("instance-a", "group-a"), 502);
+ verify(apacheProvider, never()).deleteTopic(any(), any());
}
/** Runs the resolver action against the given admin mock, wrapping failures like MqAdminExtFactory does. */
@@ -713,6 +716,7 @@ class MetadataServiceTest {
@Test
void sendMessageShouldReturnResult() {
SendMessageDTO request = SendMessageDTO.builder()
+ .instanceId("instance-a")
.topic("test-topic")
.tag("TagA")
.body("hello")
@@ -854,30 +858,6 @@ class MetadataServiceTest {
assertThat(metadataService.refreshConsumerGroup("instance-a", "cg-gone")).isNull();
}
- @Test
- void cloudConsumerGroupWriteOperationsShouldRecordServiceBoundaryAudit() {
- ConsumerGroupVO group = new ConsumerGroupVO();
- group.setName("cg-orders");
- group.setInstanceId("cloud-instance");
- group.setRetryMaxTimes(16);
- metadataService.createConsumerGroup(group);
- metadataService.updateConsumerGroup(group);
- metadataService.deleteConsumerGroup("cloud-instance", " cg-orders ");
- metadataService.resetOffset("cloud-instance", " cg-orders ", 1784246400000L, " orders ");
-
- verify(operationAuditService).record("CREATE_GROUP", "GROUP", "cg-orders",
- "cloud-instance", "consumeType=-, subscriptionMode=-, retryMaxTimes=16", "SUCCESS", null);
- verify(operationAuditService).record("UPDATE_GROUP", "GROUP", "cg-orders",
- "cloud-instance", "consumeType=-, subscriptionMode=-, retryMaxTimes=16", "SUCCESS", null);
- verify(operationAuditService).record("DELETE_GROUP", "GROUP", "cg-orders",
- "cloud-instance", null, "SUCCESS", null);
- verify(operationAuditService).record("DELETE_TOPIC", "TOPIC", "%DLQ%cg-orders",
- "cloud-instance", null, "SUCCESS", null);
- verify(operationAuditService).record("RESET_OFFSET", "GROUP", "cg-orders",
- "cloud-instance", "topic=orders, timestamp=1784246400000", "SUCCESS", null);
- verifyNoMoreInteractions(operationAuditService);
- verify(cloudProvider).resetOffset("cloud-instance", "cg-orders", 1784246400000L, "orders");
- }
@Test
void resetOffsetShouldRejectBlankTopicBeforeProviderResolution() {
@@ -940,8 +920,8 @@ class MetadataServiceTest {
}
@Test
- void importConsumerGroupsShouldContinueAfterRowFailure() {
- when(apacheProvider.createConsumerGroup(eq("instance-a"), any(ConsumerGroupVO.class)))
+ void importConsumerGroupsShouldContinueAfterRowFailureTest() {
+ when(apacheProvider.importConsumerGroup(eq("instance-a"), any(ConsumerGroupVO.class)))
.thenAnswer(invocation -> {
ConsumerGroupVO group = invocation.getArgument(1);
if ("cg-fail".equals(group.getName())) {
@@ -964,7 +944,8 @@ class MetadataServiceTest {
ArgumentCaptor<ConsumerGroupVO> captor = ArgumentCaptor.forClass(ConsumerGroupVO.class);
verify(apacheProvider, org.mockito.Mockito.times(2))
- .createConsumerGroup(eq("instance-a"), captor.capture());
+ .importConsumerGroup(eq("instance-a"), captor.capture());
+ verify(apacheProvider, never()).createConsumerGroup(anyString(), any());
assertThat(captor.getAllValues()).extracting(ConsumerGroupVO::getInstanceId)
.containsExactly("instance-a", "instance-a");
}
diff --git a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
index 929d0209f..6703bea0d 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
@@ -345,19 +345,36 @@ class TopicControllerTest extends WebMvcAuthTestSupport {
void deleteTopicShouldReturnSuccess() throws Exception {
mockMvc.perform(post("/api/topics/delete")
.contentType(MediaType.APPLICATION_JSON)
- .content(objectMapper.writeValueAsString(Map.of("name", "test-topic"))))
+ .content(objectMapper.writeValueAsString(Map.of("instanceId", "instance-a", "name", "test-topic"))))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.message").value("success"));
- verify(metadataService).deleteTopic(isNull(), eq("test-topic"));
+ verify(metadataService).deleteTopic(eq("instance-a"), eq("test-topic"));
+ }
+
+ @Test
+ void deleteTopicRequiresInstanceTest() throws Exception {
+ mockMvc.perform(post("/api/topics/delete").contentType(MediaType.APPLICATION_JSON)
+ .content("{\"name\":\"orders\"}"))
+ .andExpect(status().isBadRequest()).andExpect(jsonPath("$.code").value(400));
+ verifyNoInteractions(metadataService);
+ }
+
+ @Test
+ void deleteTopicReportsOwnershipConflictTest() throws Exception {
+ org.mockito.Mockito.doThrow(new org.apache.rocketmq.studio.common.exception.BusinessException(409, "Ownership conflict"))
+ .when(metadataService).deleteTopic("instance-a", "orders");
+ mockMvc.perform(post("/api/topics/delete").contentType(MediaType.APPLICATION_JSON)
+ .content("{\"instanceId\":\"instance-a\",\"name\":\"orders\"}"))
+ .andExpect(status().isConflict()).andExpect(jsonPath("$.code").value(409));
}
@Test
void deleteTopicShouldRejectMissingName() throws Exception {
mockMvc.perform(post("/api/topics/delete")
.contentType(MediaType.APPLICATION_JSON)
- .content("{}"))
+ .content("{\"instanceId\":\"instance-a\"}"))
.andExpect(status().isBadRequest())
.andExpect(jsonPath("$.code").value(400))
.andExpect(jsonPath("$.message").value("name is required"));
@@ -369,7 +386,7 @@ class TopicControllerTest extends WebMvcAuthTestSupport {
void deleteTopicShouldRejectBlankName() throws Exception {
mockMvc.perform(post("/api/topics/delete")
.contentType(MediaType.APPLICATION_JSON)
- .content(objectMapper.writeValueAsString(Map.of("name", " "))))
+ .content(objectMapper.writeValueAsString(Map.of("instanceId", "instance-a", "name", " "))))
.andExpect(status().isBadRequest())
.andExpect(jsonPath("$.code").value(400))
.andExpect(jsonPath("$.message").value("name is required"));
diff --git a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolInstanceRoutingTest.java b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolInstanceRoutingTest.java
index b30b997e4..8d1bdb8a4 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolInstanceRoutingTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolInstanceRoutingTest.java
@@ -87,7 +87,8 @@ class ToolInstanceRoutingTest {
AdminClient globalAdmin = mock(AdminClient.class);
MetadataService metadata = new MetadataService(globalMetadata, globalAdmin, registry, targets,
mock(OperationAuditService.class), mock(MessageService.class),
- mock(RuntimeAdminClientResolver.class));
+ mock(RuntimeAdminClientResolver.class),
+ mock(org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.class));
TopicListToolHandler handler = new TopicListToolHandler(metadata);
ToolDefinition definition = new ToolCatalog(new DefaultResourceLoader()).getDefinition(handler.name());
ToolCatalog catalog = mock(ToolCatalog.class);
diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
index 49910eb3e..2c782966c 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
@@ -34,6 +34,8 @@ import org.mockito.junit.jupiter.MockitoExtension;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
@@ -74,6 +76,33 @@ class ApacheInstanceProviderTest {
verifyNoMoreInteractions(adminClient);
}
+ @Test
+ void groupWritesRequireRequestAndExplicitCanonicalInstanceTest() {
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setInstanceId("other-instance");
+ provider.createConsumerGroup(" selected ", group);
+ assertThat(group.getInstanceId()).isEqualTo("selected");
+ provider.importConsumerGroup(" selected ", group);
+ provider.updateConsumerGroup(" selected ", group);
+ verify(adminClient).createConsumerGroup(group);
+ verify(adminClient).importConsumerGroup(group);
+ verify(adminClient).updateConsumerGroup(group);
+ }
+
+ @Test
+ void groupWritesRejectMissingInstanceOrRequestBeforeAdminTest() {
+ for (String instanceId : new String[] {null, " ", "selected"}) {
+ for (java.util.function.BiFunction<String, ConsumerGroupVO, ConsumerGroupVO> mutation :
+ java.util.List.<java.util.function.BiFunction<String, ConsumerGroupVO, ConsumerGroupVO>>of(
+ provider::createConsumerGroup, provider::importConsumerGroup, provider::updateConsumerGroup)) {
+ assertThatThrownBy(() -> mutation.apply(instanceId, null))
+ .isInstanceOfSatisfying(org.apache.rocketmq.studio.common.exception.BusinessException.class,
+ failure -> assertThat(failure.getCode()).isEqualTo(400));
+ }
+ }
+ verifyNoInteractions(adminClient);
+ }
+
@Test
void capabilitiesShouldIncludeApacheOnlyOperationsTest() {
assertThat(provider.capabilities()).contains(
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 8fa9262fe..6d6431599 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
@@ -7,6 +7,12 @@
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
*/
package org.apache.rocketmq.studio.provider.apache;
@@ -112,10 +118,41 @@ class RocketMQAdminClientImplTest {
@Mock
private DefaultMQProducer sendProducer;
+ @Mock
+ private org.apache.rocketmq.studio.instance.ResourceOwnershipGuard ownershipGuard;
+
private RocketMQAdminClientImpl adminClient;
@BeforeEach
- void setUp() {
+ void setUp() throws Exception {
+ lenient().when(ownershipGuard.requireInstance(any())).thenAnswer(call -> {
+ String name = org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.requireText(
+ call.getArgument(0), "instanceId");
+ var instance = org.apache.rocketmq.studio.instance.InstanceVO.builder().name(name).build();
+ if (!name.startsWith("cluster-") && !name.equals("configured-target")) {
+ instance.setId(1L);
+ }
+ return instance;
+ });
+ lenient().when(ownershipGuard.topicResource(any())).thenCallRealMethod();
+ lenient().when(ownershipGuard.check(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);
+ });
+ 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(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");
+ lenient().when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ lenient().when(adminExt.examineTopicRouteInfo(anyString())).thenReturn(writeRoute());
lenient().when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
lenient().when(adminFactory.execute(anyString(), any(), any())).thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(2).apply(adminExt));
@@ -124,7 +161,16 @@ class RocketMQAdminClientImplTest {
lenient().when(runtimeAdminClientResolver.executeProducer(any(), any())).thenAnswer(invocation ->
invocation.<MqClientPool.ClientAction<DefaultMQProducer, Object>>getArgument(1).apply(sendProducer));
adminClient = new RocketMQAdminClientImpl(adminFactory, properties, topicMapper, groupMapper, auditService,
- runtimeAdminClientResolver, clientPool);
+ runtimeAdminClientResolver, clientPool, ownershipGuard);
+ }
+
+ private TopicRouteData writeRoute() {
+ TopicRouteData route = new TopicRouteData();
+ route.setBrokerDatas(List.copyOf(clusterInfoWithMaster().getBrokerAddrTable().values()));
+ QueueData queue = new QueueData();
+ queue.setBrokerName("broker-1");
+ route.setQueueDatas(List.of(queue));
+ return route;
}
private String singleClusterTarget() {
@@ -360,6 +406,8 @@ class RocketMQAdminClientImplTest {
return action.apply(selectedAdmin);
});
long timestamp = 1784246400000L;
+ when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ when(selectedAdmin.examineTopicRouteInfo("orders")).thenReturn(writeRoute());
adminClient.resetOffset("instance-a", "cg-orders", timestamp, "orders");
@@ -369,14 +417,10 @@ class RocketMQAdminClientImplTest {
}
@Test
- void resetOffsetWithoutInstanceShouldUseForceAndOfflineFallbackSemantics() throws Exception {
- long timestamp = 1784246400000L;
-
- adminClient.resetOffset(null, "cg-orders", timestamp, "orders");
-
- verify(adminExt).resetOffsetNew("cg-orders", "orders", timestamp);
- verify(adminExt, never()).resetOffsetByTimestamp(
- anyString(), anyString(), anyString(), anyLong(), anyBoolean());
+ void resetOffsetWithoutInstanceRejectsBeforeRemoteTest() {
+ assertThatThrownBy(() -> adminClient.resetOffset(null, "cg-orders", 1784246400000L, "orders"))
+ .isInstanceOfSatisfying(BusinessException.class, error -> assertThat(error.getCode()).isEqualTo(400));
+ verifyNoInteractions(adminExt, runtimeAdminClientResolver);
}
@Test
@@ -402,7 +446,7 @@ class RocketMQAdminClientImplTest {
when(adminExt.searchOffset("10.0.0.2:10911", "orders", 1, timestamp, 3_000L)).thenReturn(170L);
ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
- null, "cg-orders", timestamp, "orders");
+ "instance-a", "cg-orders", timestamp, "orders");
assertThat(preview.isComplete()).isTrue();
assertThat(preview.isAllowReset()).isTrue();
@@ -443,7 +487,7 @@ class RocketMQAdminClientImplTest {
when(adminExt.searchOffset("10.0.0.1:10911", "orders", 0, timestamp, 3_000L)).thenReturn(80L);
ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
- null, "cg-orders", timestamp, "orders");
+ "instance-a", "cg-orders", timestamp, "orders");
assertThat(preview.getCurrentTotalLag()).isEqualTo(ConsumerLagResolver.UNKNOWN);
assertThat(preview.getProjectedTotalLag()).isEqualTo(20L);
@@ -470,7 +514,7 @@ class RocketMQAdminClientImplTest {
when(adminExt.searchOffset("10.0.0.1:10911", "orders", 0, timestamp, 3_000L)).thenReturn(120L);
ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
- null, "cg-orders", timestamp, "orders");
+ "instance-a", "cg-orders", timestamp, "orders");
assertThat(preview.getCurrentTotalLag()).isEqualTo(20L);
assertThat(preview.getProjectedTotalLag()).isEqualTo(ConsumerLagResolver.UNKNOWN);
@@ -490,7 +534,7 @@ class RocketMQAdminClientImplTest {
when(adminExt.minOffset(queue)).thenThrow(new IllegalStateException("offset unavailable"));
ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
- null, "cg-orders", timestamp, "orders");
+ "instance-a", "cg-orders", timestamp, "orders");
assertThat(preview.isComplete()).isFalse();
assertThat(preview.isAllowReset()).isFalse();
@@ -514,7 +558,7 @@ class RocketMQAdminClientImplTest {
when(adminExt.examineConsumeStats("cg-orders")).thenReturn(stats);
ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
- null, "cg-orders", 1784246400000L, "orders");
+ "instance-a", "cg-orders", 1784246400000L, "orders");
assertThat(preview.isComplete()).isFalse();
assertThat(preview.isAllowReset()).isFalse();
@@ -531,7 +575,7 @@ class RocketMQAdminClientImplTest {
"No topic route info in name server for the topic: %RETRY%cg-orders"));
ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
- null, "cg-orders", 1784246400000L, "orders");
+ "instance-a", "cg-orders", 1784246400000L, "orders");
assertThat(preview.isComplete()).isFalse();
assertThat(preview.isAllowReset()).isFalse();
@@ -552,7 +596,7 @@ class RocketMQAdminClientImplTest {
+ "the consumer is under the broadcast mode"));
ResetConsumerOffsetPreviewVO preview = adminClient.previewResetOffset(
- null, "cg-orders", 1784246400000L, "orders");
+ "instance-a", "cg-orders", 1784246400000L, "orders");
assertThat(preview.isComplete()).isFalse();
assertThat(preview.isAllowReset()).isFalse();
@@ -644,6 +688,7 @@ class RocketMQAdminClientImplTest {
clusterInfo.setClusterAddrTable(clusterAddrTable);
Map<String, BrokerData> brokerAddrTable = new HashMap<>();
BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("cluster-1");
brokerData.setBrokerName("broker-1");
brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
brokerAddrTable.put("broker-1", brokerData);
@@ -653,6 +698,7 @@ class RocketMQAdminClientImplTest {
doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(), any(TopicConfig.class));
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("topicA");
adminClient.createTopic(topic);
@@ -660,7 +706,7 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(topicMapper, times(2)).selectOne(captor.capture());
for (LambdaQueryWrapper<RmqTopic> wrapper : captor.getAllValues()) {
- assertThat(wrapper.getSqlSegment()).contains("cluster_id", "instance_id");
+ assertThat(wrapper.getSqlSegment()).contains("name").doesNotContain("cluster_id", "instance_id");
}
}
@@ -673,76 +719,72 @@ class RocketMQAdminClientImplTest {
doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(), any(TopicConfig.class));
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
- topic.setClusterId("cluster-1");
+ topic.setInstanceId("cluster-2");
adminClient.createTopic(topic);
- verify(adminExt).createAndUpdateTopicConfig(
- org.mockito.ArgumentMatchers.eq("10.0.0.1:10911"), any(TopicConfig.class));
- verify(adminExt, never()).createAndUpdateTopicConfig(
- org.mockito.ArgumentMatchers.eq("10.0.1.1:10911"), any(TopicConfig.class));
+ verify(adminExt).createAndUpdateTopicConfig(eq("10.0.1.1:10911"), any(TopicConfig.class));
+ verify(adminExt, never()).createAndUpdateTopicConfig(eq("10.0.0.1:10911"), any(TopicConfig.class));
}
@Test
- void createTopicFallsBackToAllBrokersWhenClusterTopologyIsMissing() throws Exception {
+ void createTopicRejectsMissingTopologyWithoutBrokerFallbackTest() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), RmqTopic.class);
ClusterInfo clusterInfo = new ClusterInfo();
Map<String, BrokerData> brokerAddrTable = new HashMap<>();
brokerAddrTable.put("missing-broker", null);
BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("cluster-1");
brokerData.setBrokerName("broker-1");
brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
brokerAddrTable.put("broker-1", brokerData);
clusterInfo.setBrokerAddrTable(brokerAddrTable);
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo);
- when(topicMapper.selectOne(any())).thenReturn(null);
-
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
-
- adminClient.createTopic(topic);
-
- verify(adminExt).createAndUpdateTopicConfig(eq("10.0.0.1:10911"), any(TopicConfig.class));
+ assertThatThrownBy(() -> adminClient.createTopic(topic))
+ .isInstanceOfSatisfying(BusinessException.class, error -> assertThat(error.getCode()).isEqualTo(409));
+ verify(adminExt, never()).createAndUpdateTopicConfig(anyString(), any(TopicConfig.class));
+ verifyNoInteractions(topicMapper);
}
@Test
- void configuredGroupCreationKeepsDefaultClusterSelectionAndEmptyMetadataScope() throws Exception {
+ void virtualGroupCreationRejectsUnmatchedPhysicalClusterTest() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), RmqGroup.class);
DefaultMQAdminExt selectedAdmin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
- when(runtimeAdminClientResolver.configuredClusterName("configured-target")).thenReturn("configured-target");
when(runtimeAdminClientResolver.execute(eq("configured-target"), any()))
.thenAnswer(invocation -> invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1)
.apply(selectedAdmin));
ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setInstanceId("instance-a");
group.setName("orders-consumers");
group.setInstanceId("configured-target");
- adminClient.createConsumerGroup(group);
- ArgumentCaptor<RmqGroup> saved = ArgumentCaptor.forClass(RmqGroup.class);
- verify(groupMapper).insert(saved.capture());
- assertThat(saved.getValue().getInstanceId()).isEmpty();
- assertThat(saved.getValue().getClusterId()).isEqualTo("cluster-1");
+ assertThatThrownBy(() -> adminClient.createConsumerGroup(group))
+ .isInstanceOfSatisfying(BusinessException.class, error -> assertThat(error.getCode()).isEqualTo(409));
+ verify(selectedAdmin, never()).createAndUpdateSubscriptionGroupConfig(anyString(), any());
+ verifyNoInteractions(groupMapper);
verifyNoInteractions(adminExt, adminFactory);
}
@Test
- void configuredTopicCreationKeepsDefaultClusterSelectionAndEmptyMetadataScope() throws Exception {
+ void virtualTopicCreationRejectsUnmatchedPhysicalClusterTest() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), RmqTopic.class);
DefaultMQAdminExt selectedAdmin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
- when(runtimeAdminClientResolver.configuredClusterName("configured-target")).thenReturn("configured-target");
when(runtimeAdminClientResolver.execute(eq("configured-target"), any()))
.thenAnswer(invocation -> invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1)
.apply(selectedAdmin));
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
- adminClient.createTopic("configured-target", topic);
- ArgumentCaptor<RmqTopic> saved = ArgumentCaptor.forClass(RmqTopic.class);
- verify(topicMapper).insert(saved.capture());
- assertThat(saved.getValue().getInstanceId()).isEmpty();
- assertThat(saved.getValue().getClusterId()).isEqualTo("cluster-1");
- verify(selectedAdmin).createAndUpdateTopicConfig(eq("10.0.0.1:10911"), any(TopicConfig.class));
+ assertThatThrownBy(() -> adminClient.createTopic("configured-target", topic))
+ .isInstanceOfSatisfying(BusinessException.class, error -> assertThat(error.getCode()).isEqualTo(409));
+ verify(selectedAdmin, never()).createAndUpdateTopicConfig(anyString(), any());
+ verifyNoInteractions(topicMapper);
verifyNoInteractions(adminExt, adminFactory);
}
@@ -759,6 +801,7 @@ class RocketMQAdminClientImplTest {
existing.setInstanceId("selected-instance");
when(topicMapper.selectOne(any())).thenReturn(create ? null : existing);
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
topic.setInstanceId("other-instance");
@@ -772,9 +815,8 @@ 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("instance_id", "cluster_id", "name");
- assertThat(query.getParamNameValuePairs().values())
- .contains("selected-instance", "cluster-1", "orders").doesNotContain("other-instance");
+ assertThat(query.getSqlSegment()).contains("name").doesNotContain("instance_id", "cluster_id");
+ assertThat(query.getParamNameValuePairs().values()).containsExactly("orders");
}
if (create) {
ArgumentCaptor<RmqTopic> saved = ArgumentCaptor.forClass(RmqTopic.class);
@@ -802,6 +844,7 @@ class RocketMQAdminClientImplTest {
.apply(selectedAdmin));
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("topicA");
topic.setInstanceId("open-source-local");
@@ -821,6 +864,7 @@ class RocketMQAdminClientImplTest {
when(topicMapper.selectOne(any())).thenReturn(null);
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("missing-topic");
assertThatThrownBy(() -> adminClient.updateTopic(topic))
@@ -833,15 +877,16 @@ class RocketMQAdminClientImplTest {
}
@Test
- void updateTopicPersistsTypeAndRemark() throws Exception {
+ void updateTopicPreservesTypeAndPersistsRemarkTest() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), RmqTopic.class);
RmqTopic existing = new RmqTopic();
- existing.setTopicType(TopicType.NORMAL.name());
+ existing.setTopicType(TopicType.FIFO.name());
existing.setRemark("old remark");
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
when(topicMapper.selectOne(any())).thenReturn(existing);
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
topic.setType(TopicType.FIFO);
topic.setRemark("updated remark");
@@ -867,6 +912,7 @@ class RocketMQAdminClientImplTest {
when(topicMapper.update(isNull(), any(UpdateWrapper.class))).thenReturn(1);
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
topic.setRemark("");
@@ -894,6 +940,7 @@ class RocketMQAdminClientImplTest {
when(topicMapper.selectOne(any())).thenReturn(existing);
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
TopicVO updated = adminClient.updateTopic(topic);
@@ -915,6 +962,7 @@ class RocketMQAdminClientImplTest {
doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(), any(TopicConfig.class));
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
topic.setType(TopicType.FIFO);
@@ -936,6 +984,7 @@ class RocketMQAdminClientImplTest {
when(topicMapper.selectOne(any())).thenReturn(existing);
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
adminClient.updateTopic(topic);
@@ -959,6 +1008,7 @@ class RocketMQAdminClientImplTest {
doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(), any(TopicConfig.class));
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
topic.setPerm(TopicPerm.RO);
@@ -985,6 +1035,7 @@ class RocketMQAdminClientImplTest {
doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(), any(TopicConfig.class));
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("orders");
topic.setWriteQueues(16);
topic.setReadQueues(12);
@@ -1017,7 +1068,7 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor<LambdaQueryWrapper<RmqTopic>> captor = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(topicMapper).delete(captor.capture());
- assertThat(captor.getValue().getSqlSegment()).contains("cluster_id", "instance_id", "name");
+ assertThat(captor.getValue().getSqlSegment()).contains("name").doesNotContain("cluster_id", "instance_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");
}
@@ -1046,6 +1097,7 @@ class RocketMQAdminClientImplTest {
ClusterInfo clusterInfo = new ClusterInfo();
clusterInfo.setClusterAddrTable(new HashMap<>(Map.of("cluster-1", new HashSet<>(List.of("broker-1")))));
BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("cluster-1");
brokerData.setBrokerName("broker-1");
brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-1", brokerData)));
@@ -1059,6 +1111,7 @@ class RocketMQAdminClientImplTest {
});
ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setInstanceId("instance-a");
group.setName("cg-orders");
group.setInstanceId("open-source-local");
@@ -1070,7 +1123,7 @@ 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("cluster_id", "instance_id", "name");
+ assertThat(captor.getValue().getSqlSegment()).contains("name").doesNotContain("cluster_id", "instance_id");
}
@ParameterizedTest
@@ -1108,6 +1161,7 @@ class RocketMQAdminClientImplTest {
entity.setId(7L);
when(groupMapper.selectOne(any())).thenReturn(entity);
ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setInstanceId("instance-a");
group.setName("cg-orders");
group.setInstanceId("instance-a");
group.setRetryMaxTimes(retryMaxTimes);
@@ -1146,6 +1200,7 @@ class RocketMQAdminClientImplTest {
reads.thenThrow(new IllegalStateException("broker unavailable"));
}
ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setInstanceId("instance-a");
group.setName("cg-orders");
group.setRetryMaxTimes(8);
@@ -1171,6 +1226,7 @@ class RocketMQAdminClientImplTest {
doNothing().doThrow(new IllegalStateException("broker unavailable")).when(adminExt)
.createAndUpdateSubscriptionGroupConfig(anyString(), any());
ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setInstanceId("instance-a");
group.setName("cg-orders");
group.setRetryMaxTimes(8);
@@ -1188,12 +1244,13 @@ class RocketMQAdminClientImplTest {
void updateConsumerGroupFailsWhenNoBrokerIsAvailable() throws Exception {
when(adminExt.examineBrokerClusterInfo()).thenReturn(new ClusterInfo());
ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setInstanceId("instance-a");
group.setName("cg-orders");
group.setRetryMaxTimes(8);
assertThatThrownBy(() -> adminClient.updateConsumerGroup(group))
.isInstanceOf(BusinessException.class)
- .hasMessageContaining("No broker available to update consumer group");
+ .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(409));
verify(adminExt, never()).createAndUpdateSubscriptionGroupConfig(anyString(), any());
verifyNoInteractions(groupMapper);
@@ -1208,7 +1265,7 @@ class RocketMQAdminClientImplTest {
.thenReturn(config)
.thenThrow(new IllegalStateException("broker unavailable"));
- assertThatThrownBy(() -> adminClient.updateConsumerGroupSettings(null, "cg-orders",
+ assertThatThrownBy(() -> adminClient.updateConsumerGroupSettings("instance-a", "cg-orders",
new ConsumerGroupSettingsCommand(2, 8, null, null, null)))
.isInstanceOf(BusinessException.class)
.hasMessageContaining("broker unavailable");
@@ -1226,7 +1283,7 @@ class RocketMQAdminClientImplTest {
.thenReturn(config)
.thenReturn(null);
- assertThatThrownBy(() -> adminClient.updateConsumerGroupSettings(null, "cg-orders",
+ assertThatThrownBy(() -> adminClient.updateConsumerGroupSettings("instance-a", "cg-orders",
new ConsumerGroupSettingsCommand(2, 8, null, null, null)))
.isInstanceOf(BusinessException.class)
.hasMessageContaining("Consumer group not found");
@@ -1242,6 +1299,7 @@ class RocketMQAdminClientImplTest {
ClusterInfo clusterInfo = new ClusterInfo();
clusterInfo.setClusterAddrTable(new HashMap<>(Map.of("cluster-1", new HashSet<>(List.of("broker-1")))));
BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("cluster-1");
brokerData.setBrokerName("broker-1");
brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-1", brokerData)));
@@ -1280,6 +1338,7 @@ class RocketMQAdminClientImplTest {
ClusterInfo clusterInfo = new ClusterInfo();
clusterInfo.setClusterAddrTable(new HashMap<>(Map.of("cluster-1", new HashSet<>(List.of("broker-1")))));
BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("cluster-1");
brokerData.setBrokerName("broker-1");
brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-1", brokerData)));
@@ -1322,6 +1381,7 @@ class RocketMQAdminClientImplTest {
ClusterInfo clusterInfo = new ClusterInfo();
clusterInfo.setClusterAddrTable(new HashMap<>(Map.of("cluster-1", new HashSet<>(List.of("broker-1")))));
BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("cluster-1");
brokerData.setBrokerName("broker-1");
brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-1", brokerData)));
@@ -1356,6 +1416,7 @@ class RocketMQAdminClientImplTest {
ClusterInfo clusterInfo = new ClusterInfo();
clusterInfo.setClusterAddrTable(new HashMap<>(Map.of("cluster-1", Set.of("broker-1"))));
BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("cluster-1");
brokerData.setBrokerName("broker-1");
brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-1", brokerData)));
@@ -1374,7 +1435,7 @@ class RocketMQAdminClientImplTest {
verify(adminExt, never()).deleteSubscriptionGroup(anyString(), anyString(), org.mockito.ArgumentMatchers.anyBoolean());
ArgumentCaptor<LambdaQueryWrapper<RmqGroup>> captor = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
verify(groupMapper).delete(captor.capture());
- assertThat(captor.getValue().getSqlSegment()).contains("cluster_id", "instance_id", "name");
+ assertThat(captor.getValue().getSqlSegment()).contains("name").doesNotContain("cluster_id", "instance_id");
}
@Test
@@ -1409,6 +1470,7 @@ class RocketMQAdminClientImplTest {
.record(any(), any(), any(), any(), any(), any());
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("topicA");
assertThat(adminClient.createTopic(topic).getId()).isEqualTo(1L);
@@ -1418,12 +1480,13 @@ class RocketMQAdminClientImplTest {
@Test
void createTopicPreservesRemoteFailureWhenAuditRecordingFails() throws Exception {
- when(adminExt.examineBrokerClusterInfo())
- .thenThrow(new IllegalStateException("broker unavailable"));
+ doThrow(new IllegalStateException("broker unavailable")).when(adminExt)
+ .createAndUpdateTopicConfig(anyString(), any());
doThrow(new RuntimeException("audit db down")).when(auditService)
.record(any(), any(), any(), any(), any(), any());
TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
topic.setName("topicA");
assertThatThrownBy(() -> adminClient.createTopic(topic))
@@ -1438,6 +1501,7 @@ class RocketMQAdminClientImplTest {
clusterInfo.setClusterAddrTable(clusterAddrTable);
Map<String, BrokerData> brokerAddrTable = new HashMap<>();
BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("cluster-1");
brokerData.setBrokerName("broker-1");
brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
brokerAddrTable.put("broker-1", brokerData);
@@ -1449,6 +1513,7 @@ class RocketMQAdminClientImplTest {
ClusterInfo clusterInfo = clusterInfoWithMaster();
clusterInfo.getClusterAddrTable().get("cluster-1").add("broker-2");
BrokerData broker = new BrokerData();
+ broker.setCluster("cluster-1");
broker.setBrokerName("broker-2");
broker.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.2:10911")));
clusterInfo.getBrokerAddrTable().put("broker-2", broker);
@@ -1463,9 +1528,11 @@ class RocketMQAdminClientImplTest {
clusterInfo.setClusterAddrTable(clusterAddrTable);
Map<String, BrokerData> brokerAddrTable = new HashMap<>();
BrokerData firstBroker = new BrokerData();
+ firstBroker.setCluster("cluster-1");
firstBroker.setBrokerName("broker-1");
firstBroker.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
BrokerData secondBroker = new BrokerData();
+ secondBroker.setCluster("cluster-2");
secondBroker.setBrokerName("broker-2");
secondBroker.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.1.1:10911")));
brokerAddrTable.put("broker-1", firstBroker);
@@ -1492,6 +1559,7 @@ class RocketMQAdminClientImplTest {
when(sendProducer.send(any(Message.class))).thenReturn(sendResult);
SendMessageDTO request = new SendMessageDTO();
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setBody("hello");
SendMessageVO result = adminClient.sendMessage(request);
@@ -1503,6 +1571,7 @@ class RocketMQAdminClientImplTest {
void sendMessageShouldRejectOversizedUtf8BodyBeforeResolvingEndpointOrCreatingProducer() {
SendMessageDTO request = new SendMessageDTO();
request.setInstanceId("instance-a");
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setBody("\u754c".repeat((4 * 1024 * 1024 / 3) + 1));
@@ -1527,6 +1596,7 @@ class RocketMQAdminClientImplTest {
when(sendProducer.send(any(Message.class))).thenReturn(sendResult);
SendMessageDTO request = new SendMessageDTO();
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setBody("x".repeat(4 * 1024 * 1024));
@@ -1536,7 +1606,8 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
verify(sendProducer).send(messageCaptor.capture());
assertThat(messageCaptor.getValue().getBody()).hasSize(4 * 1024 * 1024);
- verify(clientPool).withProducer(eq("10.0.0.1:9876"), isNull(), isNull(), any());
+ verify(runtimeAdminClientResolver).executeProducer(eq("instance-a"), any());
+ verifyNoInteractions(clientPool);
}
@Test
@@ -1548,6 +1619,7 @@ class RocketMQAdminClientImplTest {
when(sendProducer.send(any(Message.class))).thenReturn(sendResult);
SendMessageDTO request = new SendMessageDTO();
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setBody("hello");
request.setInstanceId("instance-a");
@@ -1565,6 +1637,7 @@ class RocketMQAdminClientImplTest {
when(sendProducer.send(any(Message.class))).thenReturn(sendResult);
SendMessageDTO request = new SendMessageDTO();
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setBody("hello");
@@ -1581,6 +1654,7 @@ class RocketMQAdminClientImplTest {
when(sendProducer.send(any(Message.class))).thenReturn(null);
SendMessageDTO request = new SendMessageDTO();
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setBody("hello");
@@ -1613,6 +1687,7 @@ class RocketMQAdminClientImplTest {
properties.put("bizKey", "bizValue");
SendMessageDTO request = new SendMessageDTO();
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setTag("tagA");
request.setKey("keyA");
@@ -1647,6 +1722,7 @@ class RocketMQAdminClientImplTest {
.thenReturn(sendResult);
SendMessageDTO request = new SendMessageDTO();
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setBody("hello");
request.setMessageGroup("group-x");
@@ -1675,6 +1751,7 @@ class RocketMQAdminClientImplTest {
when(sendProducer.send(any(Message.class))).thenReturn(sendResult);
SendMessageDTO request = new SendMessageDTO();
+ request.setInstanceId("instance-a");
request.setTopic("TopicA");
request.setBody("hello");
request.setDeliveryTimestamp(1_900_000_000_000L);
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 e8b213d35..a6518fd88 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
@@ -33,6 +33,9 @@ import org.apache.rocketmq.remoting.protocol.body.GetLiteGroupInfoResponseBody;
import org.apache.rocketmq.remoting.protocol.body.GetParentTopicInfoResponseBody;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.model.LiteTopicQuota;
import org.apache.rocketmq.studio.model.LiteTopicSession;
@@ -72,6 +75,8 @@ class RocketMQLiteTopicProviderTest {
private MQAdminExt admin;
private RocketMQLiteTopicProvider provider;
+ private ResourceOwnershipGuard guard;
+ private RuntimeAdminClientResolver runtime;
@BeforeEach
void setUp() {
@@ -83,12 +88,23 @@ class RocketMQLiteTopicProviderTest {
}).when(factory).execute(anyString(), nullable(RPCHook.class), any());
RocketMQProperties properties = new RocketMQProperties();
properties.setNamesrvAddr(NAMESRV);
- provider = new RocketMQLiteTopicProvider(factory, properties);
+ guard = mock(ResourceOwnershipGuard.class);
+ runtime = mock(RuntimeAdminClientResolver.class);
+ 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.withOwned(any(), any(), any())).thenAnswer(call ->
+ ((java.util.function.Supplier<?>) call.getArgument(2)).get());
+ doAnswer(call -> ((MqAdminExtFactory.AdminAction<?>) call.getArgument(1)).apply(admin))
+ .when(runtime).execute(any(InstanceVO.class), any());
+ provider = new RocketMQLiteTopicProvider(factory, properties, runtime, guard);
}
@Test
void isSupportedIsFalseWithoutConfiguredNameServer() {
- assertThat(new RocketMQLiteTopicProvider(mock(MqAdminExtFactory.class), new RocketMQProperties())
+ assertThat(new RocketMQLiteTopicProvider(mock(MqAdminExtFactory.class), new RocketMQProperties(), runtime, guard)
.isSupported()).isFalse();
}
@@ -263,12 +279,34 @@ class RocketMQLiteTopicProviderTest {
ex -> assertThat(ex.getCode()).isEqualTo(400));
}
+ @Test
+ void ttlMissingInstanceAndOwnershipConflictStopBeforeClientTest() {
+ assertThatThrownBy(() -> provider.extendTTL(PARENT, 60_000L))
+ .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"));
+ assertThatThrownBy(() -> provider.extendTTL("cluster-a", PARENT, 60_000L))
+ .isInstanceOfSatisfying(BusinessException.class, ex -> assertThat(ex.getCode()).isEqualTo(409));
+ org.mockito.Mockito.verifyNoInteractions(runtime, admin);
+ }
+
+ @Test
+ void ttlRejectsIncompleteTopologyTest() throws Exception {
+ ClusterInfo incomplete = cluster(BROKER_A);
+ incomplete.getBrokerAddrTable().get(BROKER_A).setBrokerAddrs(new HashMap<>(Map.of(1L, "slave:10911")));
+ when(admin.examineBrokerClusterInfo()).thenReturn(incomplete);
+ assertThatThrownBy(() -> provider.extendTTL("cluster-a", PARENT, 60_000L))
+ .isInstanceOfSatisfying(BusinessException.class, ex -> assertThat(ex.getCode()).isEqualTo(409));
+ verify(admin, never()).createAndUpdateTopicConfig(anyString(), any());
+ }
+
@Test
void extendTtlConvertsMillisecondsAndUpdatesTheLiteParentTopic() throws Exception {
when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
when(admin.examineTopicConfig(BROKER_A, PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
- provider.extendTTL(PARENT, TimeUnit.MINUTES.toMillis(120));
+ provider.extendTTL("cluster-a", PARENT, TimeUnit.MINUTES.toMillis(120));
ArgumentCaptor<TopicConfig> captor = ArgumentCaptor.forClass(TopicConfig.class);
verify(admin).createAndUpdateTopicConfig(eq(BROKER_A), captor.capture());
@@ -284,7 +322,7 @@ class RocketMQLiteTopicProviderTest {
when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
when(admin.examineTopicConfig(BROKER_A, PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
- provider.extendTTL(PARENT, TimeUnit.DAYS.toMillis(90));
+ provider.extendTTL("cluster-a", PARENT, TimeUnit.DAYS.toMillis(90));
ArgumentCaptor<TopicConfig> captor = ArgumentCaptor.forClass(TopicConfig.class);
verify(admin).createAndUpdateTopicConfig(eq(BROKER_A), captor.capture());
@@ -298,7 +336,7 @@ class RocketMQLiteTopicProviderTest {
when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
when(admin.examineTopicConfig(BROKER_A, PARENT)).thenReturn(new TopicConfig(PARENT));
- assertThatThrownBy(() -> provider.extendTTL(PARENT, 60_000L))
+ assertThatThrownBy(() -> provider.extendTTL("cluster-a", PARENT, 60_000L))
.isInstanceOfSatisfying(BusinessException.class,
ex -> assertThat(ex.getCode()).isEqualTo(404));
verify(admin, never()).createAndUpdateTopicConfig(anyString(), any());
@@ -314,7 +352,7 @@ class RocketMQLiteTopicProviderTest {
// The extension must fail before any master is written: the alternative is a cluster
// with mixed lite.topic.expiration attributes and a console that reports success.
- assertThatThrownBy(() -> provider.extendTTL(PARENT, TimeUnit.MINUTES.toMillis(120)))
+ assertThatThrownBy(() -> provider.extendTTL("cluster-a", PARENT, TimeUnit.MINUTES.toMillis(120)))
.isInstanceOf(RemotingTimeoutException.class);
verify(admin, never()).createAndUpdateTopicConfig(anyString(), any());
}
@@ -327,7 +365,7 @@ class RocketMQLiteTopicProviderTest {
when(admin.examineTopicConfig(peerMaster, PARENT))
.thenThrow(new MQBrokerException(ResponseCode.TOPIC_NOT_EXIST, "topic not exist"));
- provider.extendTTL(PARENT, TimeUnit.MINUTES.toMillis(120));
+ provider.extendTTL("cluster-a", PARENT, TimeUnit.MINUTES.toMillis(120));
verify(admin).createAndUpdateTopicConfig(eq(BROKER_A), any(TopicConfig.class));
verify(admin, never()).createAndUpdateTopicConfig(eq(peerMaster), any());
@@ -396,11 +434,14 @@ class RocketMQLiteTopicProviderTest {
}
ClusterInfo clusterInfo = new ClusterInfo();
clusterInfo.setBrokerAddrTable(table);
+ clusterInfo.setClusterAddrTable(Map.of("cluster-a", new HashSet<>(table.keySet())));
return clusterInfo;
}
private static BrokerData broker(String masterAddress) {
BrokerData data = new BrokerData();
+ data.setCluster("cluster-a");
+ data.setBrokerName(masterAddress);
data.setBrokerAddrs(new HashMap<>(Map.of(0L, masterAddress)));
return data;
}
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 37654fec4..c7961e5f6 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
@@ -98,6 +98,9 @@ class RocketMQMessageProviderTest {
@Mock
private DefaultMQPullConsumer pullConsumer;
+ @Mock
+ private org.apache.rocketmq.studio.instance.ResourceOwnershipGuard ownershipGuard;
+
private RocketMQMessageProvider provider;
@BeforeEach
@@ -123,7 +126,7 @@ class RocketMQMessageProviderTest {
invocation.getArgument(1);
return action.apply(pullConsumer);
});
- provider = new RocketMQMessageProvider(runtimeAdminClientResolver);
+ provider = new RocketMQMessageProvider(runtimeAdminClientResolver, ownershipGuard);
}
@Test
@@ -155,22 +158,92 @@ class RocketMQMessageProviderTest {
brokerResult.setConsumeResult(CMResult.CR_SUCCESS);
brokerResult.setRemark("consumed");
brokerResult.setSpentTimeMills(12);
- when(adminExt.consumeMessageDirectly("billing", "client-a", "orders", "msg-1"))
+ MessageExt message = prepareDirectConsumption();
+ String offsetId = message.getMsgId();
+ when(adminExt.consumeMessageDirectly("billing", "client-a", "orders", offsetId))
.thenReturn(brokerResult);
+
+ var result = provider.consumeMessageDirectly(directRequest());
+
+ assertThat(result.getConsumeResult()).isEqualTo("CR_SUCCESS");
+ assertThat(result.getRemark()).isEqualTo("consumed");
+ assertThat(result.getSpentTimeMillis()).isEqualTo(12);
+ verify(adminExt).consumeMessageDirectly("billing", "client-a", "orders", offsetId);
+ }
+
+ @Test
+ void directConsumptionRejectsInvalidActualMessageWithoutRemoteWriteTest() throws Exception {
+ MessageExt message = prepareDirectConsumption();
+ for (java.net.SocketAddress address : java.util.Arrays.asList(null,
+ InetSocketAddress.createUnresolved("unknown.invalid", 10911),
+ new InetSocketAddress("10.0.0.9", 10911))) {
+ message.setStoreHost(address);
+ assertDirectConflict();
+ }
+ message.setStoreHost(new InetSocketAddress("172.30.10.100", 10911));
+ message.setTopic("other-topic");
+ assertDirectConflict();
+ message.setTopic("orders");
+ message.setMsgId("not-an-offset-id");
+ assertDirectConflict();
+ verify(adminExt, never()).consumeMessageDirectly(anyString(), anyString(), anyString(), anyString());
+ }
+
+ @Test
+ void directConsumptionRejectsForeignOwnershipBeforeResolvingClientTest() throws Exception {
+ 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)))
+ .thenThrow(new BusinessException(409, "Resource belongs to another instance"));
+ assertDirectConflict();
+ verify(runtimeAdminClientResolver, never()).execute(anyString(), any());
+ verify(adminExt, never()).consumeMessageDirectly(anyString(), anyString(), anyString(), anyString());
+ }
+
+ private void assertDirectConflict() {
+ assertThatThrownBy(() -> provider.consumeMessageDirectly(directRequest()))
+ .isInstanceOfSatisfying(BusinessException.class, error -> assertThat(error.getCode()).isEqualTo(409));
+ }
+
+ private MessageExt prepareDirectConsumption() throws Exception {
+ 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(
+ new org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Ownership(
+ 1L, "orders", "cluster-a", "cluster-a", "NORMAL"));
+ when(ownershipGuard.withOwned(any(), any(), any())).thenAnswer(call ->
+ call.<java.util.function.Supplier<Object>>getArgument(2).get());
+ ClusterInfo topology = new ClusterInfo();
+ BrokerData broker = new BrokerData("cluster-a", "broker-a",
+ new HashMap<>(Map.of(0L, "172.30.10.100:10911")));
+ topology.setBrokerAddrTable(new HashMap<>(Map.of("broker-a", broker)));
+ topology.setClusterAddrTable(new HashMap<>(Map.of("cluster-a", Set.of("broker-a"))));
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(topology);
+ TopicRouteData route = new TopicRouteData();
+ route.setBrokerDatas(List.of(broker));
+ QueueData queue = new QueueData();
+ queue.setBrokerName("broker-a");
+ route.setQueueDatas(List.of(queue));
+ when(adminExt.examineTopicRouteInfo("orders")).thenReturn(route);
+ String offsetId = "AC1E0A6400002A9F0000000000000001";
+ MessageExt message = new MessageExt();
+ message.setTopic("orders");
+ message.setMsgId(offsetId);
+ message.setStoreHost(new InetSocketAddress("172.30.10.100", 10911));
+ when(adminExt.viewMessage("orders", "msg-1")).thenReturn(message);
+ return message;
+ }
+
+ private DirectConsumeMessageDTO directRequest() {
DirectConsumeMessageDTO request = new DirectConsumeMessageDTO();
request.setInstanceId("instance-a");
request.setTopic("orders");
request.setMsgId("msg-1");
request.setConsumerGroup("billing");
request.setClientId("client-a");
-
- org.apache.rocketmq.studio.instance.message.DirectConsumeMessageResultVO result =
- provider.consumeMessageDirectly(request);
-
- assertThat(result.getConsumeResult()).isEqualTo("CR_SUCCESS");
- assertThat(result.getRemark()).isEqualTo("consumed");
- assertThat(result.getSpentTimeMillis()).isEqualTo(12);
- verify(adminExt).consumeMessageDirectly("billing", "client-a", "orders", "msg-1");
+ return request;
}
@Test
diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index 8a4d47bf0..5fb78a98c 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -591,21 +591,8 @@ class RocketMQMetadataProviderTest {
}
@Test
- void getGroupSubscriptionsShouldCreateMissingRetryTopicBeforeQueryingTest() throws Exception {
+ void getGroupSubscriptionsShouldNeverCreateRetryTopicTest() throws Exception {
DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
- when(admin.examineTopicRouteInfo("%RETRY%group-pop"))
- .thenThrow(new IllegalStateException("route not found"));
-
- org.apache.rocketmq.remoting.protocol.body.ClusterInfo clusterInfo =
- new org.apache.rocketmq.remoting.protocol.body.ClusterInfo();
- java.util.HashMap<Long, String> brokerAddrs = new java.util.HashMap<>();
- brokerAddrs.put(0L, "10.0.0.11:10911");
- Map<String, org.apache.rocketmq.remoting.protocol.route.BrokerData> brokerAddrTable =
- new java.util.HashMap<>();
- brokerAddrTable.put("broker-a", new org.apache.rocketmq.remoting.protocol.route.BrokerData(
- "cluster-a", "broker-a", brokerAddrs));
- clusterInfo.setBrokerAddrTable(brokerAddrTable);
- when(admin.examineBrokerClusterInfo()).thenReturn(clusterInfo);
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection connection =
new org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
@@ -624,24 +611,30 @@ class RocketMQMetadataProviderTest {
newLiveProvider(admin).getGroupSubscriptions(null, "group-pop");
assertThat(subscriptions).extracting(SubscriptionEntryVO::getTopic).containsExactly("TopicA");
- org.mockito.ArgumentCaptor<org.apache.rocketmq.common.TopicConfig> captor =
- org.mockito.ArgumentCaptor.forClass(org.apache.rocketmq.common.TopicConfig.class);
- verify(admin).createAndUpdateTopicConfig(eq("10.0.0.11:10911"), captor.capture());
- assertThat(captor.getValue().getTopicName()).isEqualTo("%RETRY%group-pop");
- assertThat(captor.getValue().getReadQueueNums()).isEqualTo(1);
- assertThat(captor.getValue().getWriteQueueNums()).isEqualTo(1);
+ verify(admin).examineConsumerConnectionInfo("group-pop");
+ org.mockito.Mockito.verifyNoMoreInteractions(admin);
+ }
+
+ @Test
+ void groupReadsWithoutRetryRouteHaveNoWriteSideEffectsTest() throws Exception {
+ DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ var missing = new org.apache.rocketmq.client.exception.MQClientException(
+ org.apache.rocketmq.remoting.protocol.ResponseCode.TOPIC_NOT_EXIST, "retry route missing");
+ when(admin.examineConsumerConnectionInfo("group-pop")).thenThrow(missing);
+ when(admin.examineConsumeStats("group-pop")).thenThrow(missing);
+ RocketMQMetadataProvider provider = newLiveProvider(admin);
+
+ assertThat(provider.getGroupSubscriptions(null, "group-pop")).isEmpty();
+ assertThat(provider.getGroupProgress(null, "group-pop")).isEmpty();
+
+ verify(admin).examineConsumerConnectionInfo("group-pop");
+ verify(admin).examineConsumeStats("group-pop");
+ org.mockito.Mockito.verifyNoMoreInteractions(admin);
}
@Test
void getGroupSubscriptionsShouldReturnEmptyWhenGroupOnlyConnectsViaProxyTest() throws Exception {
DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
- org.apache.rocketmq.remoting.protocol.route.TopicRouteData route =
- new org.apache.rocketmq.remoting.protocol.route.TopicRouteData();
- java.util.HashMap<Long, String> brokerAddrs = new java.util.HashMap<>();
- brokerAddrs.put(0L, "10.0.0.11:10911");
- route.setBrokerDatas(List.of(new org.apache.rocketmq.remoting.protocol.route.BrokerData(
- "cluster-a", "broker-a", brokerAddrs)));
- when(admin.examineTopicRouteInfo("%RETRY%group-proxy")).thenReturn(route);
when(admin.examineConsumerConnectionInfo("group-proxy")).thenThrow(
new org.apache.rocketmq.client.exception.MQBrokerException(
org.apache.rocketmq.remoting.protocol.ResponseCode.CONSUMER_NOT_ONLINE,
@@ -667,13 +660,6 @@ class RocketMQMetadataProviderTest {
@Test
void getGroupSubscriptionsShouldFallBackToProxyConnectionsTest() throws Exception {
DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
- org.apache.rocketmq.remoting.protocol.route.TopicRouteData route =
- new org.apache.rocketmq.remoting.protocol.route.TopicRouteData();
- java.util.HashMap<Long, String> brokerAddrs = new java.util.HashMap<>();
- brokerAddrs.put(0L, "10.0.0.11:10911");
- route.setBrokerDatas(List.of(new org.apache.rocketmq.remoting.protocol.route.BrokerData(
- "cluster-a", "broker-a", brokerAddrs)));
- when(admin.examineTopicRouteInfo("%RETRY%group-proxy")).thenReturn(route);
when(admin.examineConsumerConnectionInfo("group-proxy")).thenThrow(
new org.apache.rocketmq.client.exception.MQBrokerException(
org.apache.rocketmq.remoting.protocol.ResponseCode.CONSUMER_NOT_ONLINE,