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 9c270f50b feat(litetopic): create Lite parent topics with an explicit
TTL (#6644)
9c270f50b is described below
commit 9c270f50bbbe160309338b86bdd568ece2f5bb4f
Author: wizcraft_kris <[email protected]>
AuthorDate: Tue Oct 27 16:15:31 2026 +0800
feat(litetopic): create Lite parent topics with an explicit TTL (#6644)
Adds the create half of LiteTopic parent-topic lifecycle management.
A Lite parent topic has to carry `+message.type=LITE` and
`+lite.topic.expiration`
in the same write, or the broker ends up with a parent that never expires.
This
goes through `MetadataService.createTopic`, so ownership, audit and catalog
behaviour match every other topic create, and the delete half is left for a
follow-up.
Three points were fixed while landing it:
1. Instance-level capability gate. `guardLiteCreate` now runs after
`requireWriteInstance` (so it sees the canonical instance name) and calls
`ownershipGuard.requireInstance` + `requireSupportedProvider`, the same
pair
`extendTTL` uses. Without it a direct API call with `type=LITE` could
create a
Lite parent on a cloud-vendor instance:
`AliyunInstanceProvider.createTopic`
forwards only `messageType` and has no TTL field, so the topic would
exist
outside the vendor control plane with no expiry. Cloud vendors now get
501.
2. An omitted TTL is filled with `DEFAULT_LITE_TTL_MINUTES` (1440) instead
of
being forwarded as absent. In rocketmq-common 5.5.0,
`TopicAttributes.LITE_EXPIRATION_ATTRIBUTE` is
`LongRangeAttribute("lite.topic.expiration", true, -1,
DAYS.toMinutes(30), -1)`
- min -1, max 43200 minutes, default -1 - and a non-positive value means
the
topic never expires. The AI tool `rmq.topic.update` can also create a
topic
and its input schema has no TTL field, so without a server-side default
that
path could create a never-expiring parent. This mirrors how
`writeQueues` /
`readQueues` already fall back to 8.
3. Unit divergence documented rather than papered over. The native unit of
`lite.topic.expiration` is minutes, so create takes minutes; `extendTTL`
takes milliseconds. Converting create to milliseconds would lose
precision
(round(ms/60000)) and 30 days does not fit an int, so both units are
kept and
stated together in `docs/api-spec.md` 5.3 / 5.4 / 5.11, in the DTO and VO
javadoc, and in the front-end field hint, with the conversion bound
`minutes = clamp(round(millis / 60000), 1, 43200)`.
Contract sync: `TopicVO` exposes `liteTtlMinutes` with
`@JsonInclude(NON_NULL)`
so it is absent for non-Lite topics, and `UpdateTopicDTO` binds the field
so the
documented "400 when liteTtlMinutes is sent with a non-LITE type" is
actually
enforced instead of silently dropped.
Scope note: `liteTopicProvider.isSupported()` probes the nameserver address
from
Studio's own configuration, not the target instance's, so the 501 gate is
cluster-global - the same semantics the existing `GET
/api/liteTopic/capability`
endpoint already has. In a multi-instance deployment where the default
endpoint
and the target instance are different clusters, the gate can be wrong in
either
direction. `docs/api-spec.md` 5.3 states this scope explicitly; making the
probe
instance-aware needs `LiteTopicProvider.isSupported(instanceId)` plus a
resolver,
which touches the capability endpoint and its contract, so it is left to a
follow-up.
---
docs/api-spec.md | 21 +++
.../studio/instance/topic/CreateTopicDTO.java | 18 +++
.../studio/instance/topic/MetadataService.java | 49 +++++++
.../rocketmq/studio/instance/topic/TopicVO.java | 13 ++
.../studio/instance/topic/UpdateTopicDTO.java | 8 ++
.../studio/provider/LiteTopicProvider.java | 15 +++
.../provider/apache/RocketMQAdminClientImpl.java | 29 ++++-
.../provider/apache/RocketMQLiteTopicProvider.java | 3 -
.../studio/instance/topic/MetadataServiceTest.java | 143 +++++++++++++++++++++
.../studio/instance/topic/TopicControllerTest.java | 95 ++++++++++++++
.../ai/tool/service/ToolInstanceRoutingTest.java | 3 +-
.../apache/RocketMQAdminClientImplTest.java | 107 ++++++++++++++-
web/src/api/metadata.ts | 2 +
web/src/i18n/translations.ts | 13 ++
.../pages/instance/__tests__/TopicPage.test.tsx | 79 +++++++++++-
web/src/pages/instance/topic.tsx | 43 ++++++-
web/src/pages/studio/LiteTopic.tsx | 11 ++
web/src/pages/studio/__tests__/LiteTopic.test.tsx | 18 ++-
18 files changed, 653 insertions(+), 17 deletions(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index 622640a9a..567d20694 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -979,6 +979,7 @@ GET
/api/topics?clusterId={clusterId}&type={type}&search={keyword}
| `tps` | `number` | 当前 TPS |
| `consumerGroupCount` | `number` | 订阅消费组数 |
| `remark` | `string` | 备注 |
+| `liteTtlMinutes` | `number` | Lite 父主题 TTL（分钟）。**仅出现在创建请求与创建响应中**：列表 / 分页 /
详情等读响应不返回该键（服务端从不回填，不把未读取的值当作已知值呈现）。已创建父主题的 TTL 用 5.9 `GET /api/liteTopic/list`
的 `averageTTL`（毫秒）读取，从未写入过 TTL 的父主题该值为 `null` |
| `createdAt` | `string` | 创建时间 (ISO 8601) |
| `updatedAt` | `string` | 更新时间 (ISO 8601) |
@@ -1027,6 +1028,19 @@ POST /api/topics/create
| `readQueues` | `number` | 否 | 读队列数，须 ≥ 0；缺省或非正数时取 8 |
| `perm` | `string` | 否 | 权限，默认 `RW` |
| `remark` | `string` | 否 | 备注 |
+| `liteTtlMinutes` | `number` | 否 | Lite 父主题 TTL，**单位分钟**，仅 `type=LITE` 时接受，取值
1-43200（30 天）；缺省时服务端取 `1440`（1 天），与控制台弹窗预填值一致。与 `type` 一并作为创建时属性写入
broker（`+message.type=LITE` 与 `+lite.topic.expiration=<分钟>`），创建后既不能由本接口也不能由 5.4
修改 |
+
+**`type=LITE` 的额外校验**（全部在任何写入发生之前失败，不产生 broker 写入与审计记录）：
+
+| 场景 | `code` | `message` |
+|------|--------|-----------|
+| 非 `LITE` 类型携带 `liteTtlMinutes` | `400` | `liteTtlMinutes is only accepted
when type is LITE` |
+| `liteTtlMinutes` 越界（< 1 或 > 43200） | `400` | `liteTtlMinutes must be between
1 and 43200` |
+| 父主题名含 `$`（保留给 `%LMQ%$parent$liteTopic` 子主题） | `400` | `Lite parent topic
name must not contain '$': '$' separates the parent topic from its lite topics`
|
+| 目标实例不是 Apache 厂商（云厂商 OpenAPI 没有 `lite.topic.expiration` 等价物，建出的父主题会停在 broker
的 `-1` 永不过期默认值上） | `501` | `Cloud resources are not registered for global
ownership yet; resource writes are not allowed` |
+| 集群不支持 LiteTopic（能力探测与 5.13 同源，走 Studio 配置的默认接入点，非目标实例接入点） | `501` |
`LiteTopic is not supported by this provider` |
+
+**单位口径**：本接口的 `liteTtlMinutes` 用**分钟**（broker 属性 `lite.topic.expiration`
的原生单位，写入无损）；同一资源上 5.11 `POST /api/liteTopic/extendTTL` 的 `newTTL` 用**毫秒**，换算边界为
`分钟 = clamp(round(毫秒 / 60000), 1, 43200)`（亚分钟精度四舍五入到分钟，并夹在 1 分钟与 30 天之间）。
**Response `data`:** `Topic`
@@ -1049,6 +1063,9 @@ POST /api/topics/update
| `readQueues` | `number` | 否 | 读队列数，须 ≥ 0；缺省或非正数时保留现值 |
| `perm` | `string` | 否 | 权限。Apache：显式传值即更新权限；省略时各既有 Broker 保留自身运行时的权限掩码（含
Studio 不可编辑的位）、库内存值不变；新增 Broker 用库内存值兜底，取不到时用 `RW` |
| `remark` | `string` | 否 | 备注 |
+| `liteTtlMinutes` | `number` | 否 | **恒定拒绝**：该字段是创建专用字段，更新请求携带它一律返回
`400`（见下）。DTO 仍然绑定它，为的是给出明确错误而不是把未知属性静默丢弃 |
+
+`liteTtlMinutes` 是创建专用字段：更新请求携带该字段返回 `400 liteTtlMinutes is create-only; use
the LiteTopic TTL extension to change it`；更新路径也从不把 `lite.topic.expiration` 放进
broker 写增量（增量里只有 `+message.type`），已存在的 TTL 因此不被改写。调整 TTL 走 5.11，注意那边 `newTTL`
的单位是**毫秒**，与本接口创建时的**分钟**不同。
**Response `data`:** `Topic`
@@ -1217,6 +1234,10 @@ POST /api/liteTopic/extendTTL
| `topicPattern` | `string` | 是 | LiteTopic 模式 |
| `newTTL` | `number` | 是 | 新的 TTL（毫秒），必须为正数 |
+**单位口径**：`newTTL` 用**毫秒**，与 5.3 创建 Lite 父主题时的
`liteTtlMinutes`（**分钟**）不同——两者操作的是同一个 broker 属性 `lite.topic.expiration`。换算边界为
`分钟 = clamp(round(毫秒 / 60000), 1, 43200)`：亚分钟精度四舍五入到分钟，结果夹在 1 分钟与 30 天之间，因此
`newTTL` 小于 30000 毫秒一律落成 1 分钟、大于 2592000000 毫秒一律落成 43200 分钟。
+
+**实例门禁**：父主题只能属于 Apache 厂商实例（与 5.3 的 `type=LITE` 创建一致）；非 Apache 厂商返回
`501`，父主题未在 Studio 注册返回 `409`，父主题在目标集群上不存在返回 `404`。多 master 部分写入失败返回
`502`，`message` 中列出已改写的 master、失败的 master 与未尝试的 master；延长操作幂等，重跑即可收敛。
+
**Response `data`:** `null`
### 5.12 获取 LiteTopic 配额
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/CreateTopicDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/CreateTopicDTO.java
index 91bb64ddf..28bc175a4 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/CreateTopicDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/CreateTopicDTO.java
@@ -16,11 +16,14 @@
*/
package org.apache.rocketmq.studio.instance.topic;
+import jakarta.validation.constraints.Max;
+import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.PositiveOrZero;
import lombok.Data;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+import org.apache.rocketmq.studio.provider.LiteTopicProvider;
@Data
public class CreateTopicDTO {
@@ -36,6 +39,20 @@ public class CreateTopicDTO {
private TopicPerm perm;
private String remark;
+ /**
+ * TTL of a new Lite parent topic in minutes — only accepted with {@code
type=LITE} and only on
+ * create; the generic update path never edits it. Omitting it falls back
to
+ * {@link LiteTopicProvider#DEFAULT_LITE_TTL_MINUTES} instead of the
broker's own {@code -1}
+ * (never expire) attribute default, which is also what the console dialog
prefills.
+ *
+ * <p>Unit divergence on the same resource: this create field is in
minutes while
+ * {@code POST /api/liteTopic/extendTTL} takes {@code newTTL} in
milliseconds. Both are
+ * documented in {@code docs/api-spec.md} (5.3 and 5.11).
+ */
+ @Min(value = 1, message = "liteTtlMinutes must be between 1 and 43200")
+ @Max(value = LiteTopicProvider.MAX_LITE_TTL_MINUTES, message =
"liteTtlMinutes must be between 1 and 43200")
+ private Integer liteTtlMinutes;
+
private String instanceId;
public TopicVO toTopicVO() {
@@ -52,6 +69,7 @@ public class CreateTopicDTO {
}
topic.setPerm(perm);
topic.setRemark(remark);
+ topic.setLiteTtlMinutes(liteTtlMinutes);
return topic;
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
index 055439569..562fa384e 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
@@ -41,6 +41,7 @@ import
org.apache.rocketmq.studio.provider.apache.MetadataProvider;
import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.util.CsvUtil;
import org.apache.rocketmq.studio.common.util.MessagePropertyDisplay;
@@ -59,6 +60,7 @@ import
org.apache.rocketmq.studio.instance.message.MessageRecordVO;
import org.apache.rocketmq.studio.instance.message.MessageService;
import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
+import org.apache.rocketmq.studio.provider.LiteTopicProvider;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -91,6 +93,7 @@ public class MetadataService {
private final MessageService messageService;
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
private final ResourceOwnershipGuard ownershipGuard;
+ private final LiteTopicProvider liteTopicProvider;
/**
* Canonicalizes registered instance names and legacy numeric IDs, while
preserving physical
@@ -170,6 +173,7 @@ public class MetadataService {
}
String target = requireWriteInstance(instanceId, new
Resource(Kind.TOPIC, topic.getName()), false);
topic.setInstanceId(target);
+ guardLiteCreate(target, topic);
InstanceProvider provider = resolve(target);
return executeWithAudit(provider, Operation.CREATE_TOPIC,
ResourceType.TOPIC, topic.getName(),
target, topicDetail(topic), () -> importing
@@ -182,6 +186,10 @@ public class MetadataService {
public TopicVO updateTopic(String instanceId, TopicVO topic) {
requireTopic(topic);
+ if (topic.getLiteTtlMinutes() != null) {
+ throw new BusinessException(400,
+ "liteTtlMinutes is create-only; use the LiteTopic TTL
extension to change it");
+ }
String target = requireWriteInstance(instanceId, new
Resource(Kind.TOPIC, topic.getName()), false);
topic.setInstanceId(target);
guardImmutableType(target, topic);
@@ -202,6 +210,47 @@ public class MetadataService {
});
}
+ /**
+ * A Lite parent topic is created through the generic topic dialog. Its
TTL is the create-time
+ * {@code lite.topic.expiration} attribute in minutes, the {@code '$'}
name separator is
+ * reserved for the implicit {@code %LMQ%$parent$liteTopic} children, and
the write needs an
+ * Apache-vendor instance on a lite-capable cluster.
+ *
+ * <p>The instance gate mirrors {@link LiteTopicService#extendTTL}. A
cloud OpenAPI has no
+ * {@code lite.topic.expiration} equivalent, so a LITE parent created
there would keep the
+ * broker's own {@code -1} (never expire) default and stay outside that
vendor's control plane.
+ *
+ * <p>{@code instanceId} is the canonical name resolved by {@code
requireWriteInstance}, so the
+ * guard sees the same instance the write is about to target.
+ */
+ private void guardLiteCreate(String instanceId, TopicVO topic) {
+ if (topic.getLiteTtlMinutes() != null && topic.getType() !=
TopicType.LITE) {
+ throw new BusinessException(400, "liteTtlMinutes is only accepted
when type is LITE");
+ }
+ if (topic.getLiteTtlMinutes() != null
+ && (topic.getLiteTtlMinutes() < 1
+ || topic.getLiteTtlMinutes() >
LiteTopicProvider.MAX_LITE_TTL_MINUTES)) {
+ throw new BusinessException(400, "liteTtlMinutes must be between 1
and 43200");
+ }
+ if (topic.getType() != TopicType.LITE) {
+ return;
+ }
+ if (topic.getName().contains("$")) {
+ throw new BusinessException(400,
+ "Lite parent topic name must not contain '$': '$'
separates the parent topic from its lite topics");
+ }
+ InstanceVO instance = ownershipGuard.requireInstance(instanceId);
+ ownershipGuard.requireSupportedProvider(instance);
+ if (!liteTopicProvider.isSupported()) {
+ throw new BusinessException(LiteTopicProvider.NOT_IMPLEMENTED,
LiteTopicProvider.UNSUPPORTED);
+ }
+ if (topic.getLiteTtlMinutes() == null) {
+ // The broker treats a missing or non-positive
lite.topic.expiration as "never expire",
+ // so an omitted TTL is filled with the console default rather
than forwarded as-is.
+
topic.setLiteTtlMinutes(LiteTopicProvider.DEFAULT_LITE_TTL_MINUTES);
+ }
+ }
+
public void deleteTopic(String name) {
deleteTopic(null, name);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicVO.java
index 52a277120..a5dc28328 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicVO.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.instance.topic;
+import com.fasterxml.jackson.annotation.JsonInclude;
import org.apache.rocketmq.studio.common.domain.BaseEntity;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
@@ -37,4 +38,16 @@ public class TopicVO extends BaseEntity {
private double tps;
private int consumerGroupCount;
private String remark;
+ /**
+ * Create-only TTL for Lite parent topics in minutes ({@code type=LITE});
never populated on
+ * read paths and never edited by the generic update path. Kept out of the
JSON when unset so a
+ * list or detail response does not advertise a TTL the server never read
back.
+ *
+ * <p>Unit note: this create-time field is in <em>minutes</em>, the unit
of the broker's
+ * {@code lite.topic.expiration} attribute, while {@code POST
/api/liteTopic/extendTTL} takes
+ * {@code newTTL} in <em>milliseconds</em> and the provider rounds it with
+ * {@code clamp(round(millis / 60000), 1, 43200)}.
+ */
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private Integer liteTtlMinutes;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/UpdateTopicDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/UpdateTopicDTO.java
index 6a5da4e5a..ef949af42 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/UpdateTopicDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/UpdateTopicDTO.java
@@ -37,6 +37,13 @@ public class UpdateTopicDTO {
private TopicPerm perm;
private String remark;
+ /**
+ * Bound only so an update that carries a Lite TTL is rejected with an
actionable error instead
+ * of being dropped as an unknown property: the TTL is create-only and
changes go through
+ * {@code POST /api/liteTopic/extendTTL}. {@code
MetadataService#updateTopic} raises the 400.
+ */
+ private Integer liteTtlMinutes;
+
public TopicVO toTopicVO() {
TopicVO topic = new TopicVO();
topic.setName(name);
@@ -51,6 +58,7 @@ public class UpdateTopicDTO {
}
topic.setPerm(perm);
topic.setRemark(remark);
+ topic.setLiteTtlMinutes(liteTtlMinutes);
return topic;
}
}
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 87a976db4..d37cf869b 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
@@ -41,6 +41,21 @@ public interface LiteTopicProvider {
String UNSUPPORTED = "LiteTopic is not supported by this provider";
int NOT_IMPLEMENTED = 501;
+ /**
+ * Upper bound on a lite parent topic's TTL in minutes — mirrors
+ * {@code TopicAttributes.LITE_EXPIRATION_ATTRIBUTE}'s 30-day cap. The
value must stay a
+ * literal: bean-validation annotations reference it and require a
compile-time constant.
+ */
+ int MAX_LITE_TTL_MINUTES = 43_200;
+
+ /**
+ * TTL a new lite parent topic gets when the create request omits one, in
minutes. The broker
+ * attribute itself defaults to {@code -1}, i.e. never expire, so a create
that carries no TTL
+ * must not be forwarded as-is: it would leave an immortal parent topic
behind. One day matches
+ * the console dialog's prefilled value.
+ */
+ int DEFAULT_LITE_TTL_MINUTES = 1_440;
+
/**
* Whether the active cluster exposes the LiteTopic admin surface.
Detected by probing the
* broker; a cluster running a RocketMQ build without LiteTopic answers
the probe with an
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 ca6016f58..51e63c093 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
@@ -75,6 +75,7 @@ import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -321,7 +322,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
topicConfig.setWriteQueueNums(writeQueues);
topicConfig.setReadQueueNums(readQueues);
topicConfig.setPerm(toRocketMQPerm(effectivePerm));
- applyTopicType(topicConfig, topic.getType());
+ applyTopicType(topicConfig, topic.getType(),
topic.getLiteTtlMinutes());
boolean physicalExists = importing && verifyTopicImport(admin,
brokerAddrs, topicConfig);
if (!physicalExists) {
@@ -433,7 +434,9 @@ public class RocketMQAdminClientImpl implements AdminClient
{
}
// Broker attributes are merged separately and writes
require delta keys.
topicConfig.setAttributes(Map.of());
- applyTopicType(topicConfig, topic.getType());
+ // TTL is create-only: the update path passes none so a
stored
+ // lite.topic.expiration stays untouched (see the pin
test).
+ applyTopicType(topicConfig, topic.getType(), null);
pending.put(addr, topicConfig);
}
for (Map.Entry<String, TopicConfig> entry :
pending.entrySet()) {
@@ -509,13 +512,25 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return new TopicConfig(current);
}
- private void applyTopicType(TopicConfig topicConfig, TopicType topicType) {
- if (topicType == null) {
+ /**
+ * Writes the create-time topic attributes as one change set. The update
protocol only accepts
+ * attribute change entries ({@code +key=value}) which the broker merges
into the stored
+ * attributes, so the message type and a Lite parent topic's TTL are sent
together and only on
+ * create — updates pass no TTL and never touch {@code
lite.topic.expiration}.
+ */
+ private void applyTopicType(TopicConfig topicConfig, TopicType topicType,
Integer liteTtlMinutes) {
+ if (topicType == null && liteTtlMinutes == null) {
return;
}
- topicConfig.setAttributes(Map.of(
- "+" + TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(),
- topicType.name()));
+ Map<String, String> attributes = new HashMap<>();
+ if (topicType != null) {
+ attributes.put("+" +
TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(), topicType.name());
+ }
+ if (topicType == TopicType.LITE && liteTtlMinutes != null) {
+ attributes.put("+" +
TopicAttributes.LITE_EXPIRATION_ATTRIBUTE.getName(),
+ String.valueOf(liteTtlMinutes));
+ }
+ topicConfig.setAttributes(attributes);
}
@Override
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 478988355..82cf494be 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
@@ -97,9 +97,6 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
/** Upper bound on per-lite-topic offset lookups for one session detail. */
static final int MAX_SESSION_LITE_TOPIC_SCAN = 200;
- /** Mirror of {@code TopicAttributes.LITE_EXPIRATION_ATTRIBUTE}'s upper
bound (30 days). */
- static final int MAX_LITE_TTL_MINUTES = (int) TimeUnit.DAYS.toMinutes(30);
-
private final MqAdminExtFactory adminFactory;
private final RocketMQProperties properties;
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
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 5276ecf2f..831c3796e 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
@@ -111,6 +111,9 @@ class MetadataServiceTest {
@Mock
private ResourceOwnershipGuard ownershipGuard;
+ @Mock
+ private org.apache.rocketmq.studio.provider.LiteTopicProvider
liteTopicProvider;
+
@InjectMocks
private MetadataService metadataService;
@@ -621,6 +624,146 @@ class MetadataServiceTest {
verifyNoInteractions(apacheProvider, operationAuditService);
}
+ @Test
+ void createTopicShouldPassLiteTtlMinutesToProviderTest() {
+ lenient().when(liteTopicProvider.isSupported()).thenReturn(true);
+ TopicVO input = new TopicVO();
+ input.setName("chat");
+ input.setInstanceId("instance-a");
+ input.setType(TopicType.LITE);
+ input.setLiteTtlMinutes(1440);
+ when(apacheProvider.createTopic(any(),
any(TopicVO.class))).thenReturn(input);
+
+ TopicVO result = metadataService.createTopic(input);
+
+ assertThat(result.getLiteTtlMinutes()).isEqualTo(1440);
+ verify(apacheProvider).createTopic("instance-a", input);
+ }
+
+ @Test
+ void createTopicRejectsLiteTtlMinutesForNonLiteTypesTest() {
+ TopicVO input = new TopicVO();
+ input.setName("orders");
+ input.setInstanceId("instance-a");
+ input.setType(TopicType.NORMAL);
+ input.setLiteTtlMinutes(1440);
+
+ assertThatThrownBy(() -> metadataService.createTopic(input))
+ .isInstanceOfSatisfying(BusinessException.class, error -> {
+ assertThat(error.getCode()).isEqualTo(400);
+ assertThat(error.getMessage()).contains("liteTtlMinutes is
only accepted when type is LITE");
+ });
+
+ verifyNoInteractions(apacheProvider);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, -1, 43201})
+ void createTopicRejectsLiteTtlMinutesOutsideBoundsTest(int ttlMinutes) {
+ TopicVO input = new TopicVO();
+ input.setName("chat");
+ input.setInstanceId("instance-a");
+ input.setType(TopicType.LITE);
+ input.setLiteTtlMinutes(ttlMinutes);
+
+ assertThatThrownBy(() -> metadataService.createTopic(input))
+ .isInstanceOfSatisfying(BusinessException.class, error -> {
+ assertThat(error.getCode()).isEqualTo(400);
+ assertThat(error.getMessage()).contains("liteTtlMinutes
must be between 1 and 43200");
+ });
+
+ verifyNoInteractions(apacheProvider);
+ }
+
+ @Test
+ void createTopicRejectsDollarInLiteParentNameTest() {
+ TopicVO input = new TopicVO();
+ input.setName("chat$room");
+ input.setInstanceId("instance-a");
+ input.setType(TopicType.LITE);
+ input.setLiteTtlMinutes(1440);
+
+ assertThatThrownBy(() -> metadataService.createTopic(input))
+ .isInstanceOfSatisfying(BusinessException.class, error -> {
+ assertThat(error.getCode()).isEqualTo(400);
+ assertThat(error.getMessage()).contains("must not contain
'$'");
+ });
+
+ verifyNoInteractions(apacheProvider);
+ }
+
+ @Test
+ void createTopicRejectsLiteWhenLiteTopicIsUnsupportedTest() {
+ when(liteTopicProvider.isSupported()).thenReturn(false);
+ TopicVO input = new TopicVO();
+ input.setName("chat");
+ input.setInstanceId("instance-a");
+ input.setType(TopicType.LITE);
+ input.setLiteTtlMinutes(1440);
+
+ assertThatThrownBy(() -> metadataService.createTopic(input))
+ .isInstanceOfSatisfying(BusinessException.class, error -> {
+ assertThat(error.getCode()).isEqualTo(501);
+ assertThat(error.getMessage()).isEqualTo("LiteTopic is not
supported by this provider");
+ });
+
+ verifyNoInteractions(apacheProvider);
+ }
+
+ @Test
+ void updateTopicRejectsLiteTtlMinutesTest() {
+ TopicVO input = new TopicVO();
+ input.setName("chat");
+ input.setInstanceId("instance-a");
+ input.setType(TopicType.LITE);
+ input.setLiteTtlMinutes(60);
+
+ assertThatThrownBy(() -> metadataService.updateTopic(input))
+ .isInstanceOfSatisfying(BusinessException.class, error -> {
+ assertThat(error.getCode()).isEqualTo(400);
+ assertThat(error.getMessage()).contains("create-only");
+ });
+
+ verifyNoInteractions(apacheProvider);
+ }
+
+ @Test
+ void createTopicRejectsLiteParentOnCloudVendorInstanceTest() {
+ // A cloud OpenAPI has no lite.topic.expiration equivalent, so a LITE
parent created there
+ // would keep the broker's -1 (never expire) default and stay outside
the vendor's control
+ // plane. The gate is the same requireSupportedProvider the LiteTopic
TTL write uses.
+ TopicVO input = new TopicVO();
+ input.setName("chat");
+ input.setInstanceId("cloud-instance");
+ input.setType(TopicType.LITE);
+ input.setLiteTtlMinutes(1440);
+
+ assertThatThrownBy(() -> metadataService.createTopic(input))
+ .isInstanceOfSatisfying(BusinessException.class, error -> {
+ assertThat(error.getCode()).isEqualTo(501);
+ assertThat(error.getMessage()).contains("Cloud resources
are not registered");
+ });
+
+ verifyNoInteractions(apacheProvider, cloudProvider,
operationAuditService);
+ }
+
+ @Test
+ void createTopicDefaultsLiteTtlMinutesWhenOmittedTest() {
+ lenient().when(liteTopicProvider.isSupported()).thenReturn(true);
+ TopicVO input = new TopicVO();
+ input.setName("chat");
+ input.setInstanceId("instance-a");
+ input.setType(TopicType.LITE);
+ when(apacheProvider.createTopic(any(),
any(TopicVO.class))).thenReturn(input);
+
+ metadataService.createTopic(input);
+
+ // Pinned to the literal so raising or lowering the default fails
here: forwarding a LITE
+ // create with no TTL would store the broker's -1, i.e. a parent that
never expires.
+ assertThat(input.getLiteTtlMinutes()).isEqualTo(1440);
+ verify(apacheProvider).createTopic("instance-a", input);
+ }
+
@Test
void apacheTopicWriteOperationsShouldNotDuplicateProviderAudit() {
TopicVO topic = new TopicVO();
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 6703bea0d..702b37884 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
@@ -24,6 +24,7 @@ import org.springframework.context.annotation.Import;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -271,6 +272,75 @@ class TopicControllerTest extends WebMvcAuthTestSupport {
verifyNoInteractions(metadataService);
}
+ @Test
+ void createTopicShouldPassLiteTtlMinutesTest() throws Exception {
+ TopicVO created = new TopicVO();
+ created.setName("chat");
+ created.setLiteTtlMinutes(1440);
+
when(metadataService.createTopic(any(TopicVO.class))).thenReturn(created);
+
when(instanceService.normalizeIdentifier("1")).thenReturn("open-source-local");
+
+ mockMvc.perform(post("/api/topics/create")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content("""
+ {
+ "instanceId": "1",
+ "name": "chat",
+ "type": "LITE",
+ "writeQueues": 8,
+ "readQueues": 8,
+ "liteTtlMinutes": 1440
+ }
+ """))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.liteTtlMinutes").value(1440));
+
+ ArgumentCaptor<TopicVO> captor =
ArgumentCaptor.forClass(TopicVO.class);
+ verify(metadataService).createTopic(captor.capture());
+ assertThat(captor.getValue().getName()).isEqualTo("chat");
+ assertThat(captor.getValue().getLiteTtlMinutes()).isEqualTo(1440);
+ }
+
+ @Test
+ void createTopicShouldRejectLiteTtlMinutesOutsideBoundsTest() throws
Exception {
+ mockMvc.perform(post("/api/topics/create")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content("""
+ {
+ "name": "chat",
+ "type": "LITE",
+ "writeQueues": 8,
+ "readQueues": 8,
+ "liteTtlMinutes": 43201
+ }
+ """))
+ .andExpect(status().isBadRequest())
+ .andExpect(jsonPath("$.code").value(400))
+ .andExpect(jsonPath("$.message").value("liteTtlMinutes must be
between 1 and 43200"));
+
+ verifyNoInteractions(metadataService);
+ }
+
+ @Test
+ void topicListResponseOmitsTheCreateOnlyLiteTtlTest() throws Exception {
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+ topic.setType(TopicType.LITE);
+ topic.setWriteQueues(8);
+ topic.setReadQueues(8);
+ when(metadataService.listTopics(isNull(), isNull(), isNull(),
isNull())).thenReturn(List.of(topic));
+
+ // A read path never computes the Lite TTL, so the key must not be
advertised at all. The
+ // assertion is on the raw body because jsonPath(...).doesNotExist()
also passes for an
+ // explicit JSON null, which is exactly the shape this guards against.
+ String body = mockMvc.perform(get("/api/topics"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data[0].name").value("orders"))
+ .andReturn().getResponse().getContentAsString();
+
+ assertThat(body).doesNotContain("liteTtlMinutes");
+ }
+
@Test
void topicWriteEndpointsShouldRejectNullRequestBody() throws Exception {
mockMvc.perform(post("/api/topics/create")
@@ -316,6 +386,31 @@ class TopicControllerTest extends WebMvcAuthTestSupport {
verifyNoInteractions(metadataService);
}
+ @Test
+ void updateTopicBindsLiteTtlMinutesSoTheServiceCanRejectItTest() throws
Exception {
+ TopicVO updated = new TopicVO();
+ updated.setName("chat");
+
when(metadataService.updateTopic(any(TopicVO.class))).thenReturn(updated);
+
+ mockMvc.perform(post("/api/topics/update")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content("""
+ {
+ "instanceId": "1",
+ "name": "chat",
+ "type": "LITE",
+ "liteTtlMinutes": 60
+ }
+ """))
+ .andExpect(status().isOk());
+
+ // UpdateTopicDTO binds the create-only field on purpose:
MetadataService turns it into an
+ // explicit 400 instead of the request being accepted with the TTL
silently dropped.
+ ArgumentCaptor<TopicVO> captor =
ArgumentCaptor.forClass(TopicVO.class);
+ verify(metadataService).updateTopic(captor.capture());
+ assertThat(captor.getValue().getLiteTtlMinutes()).isEqualTo(60);
+ }
+
@Test
void sendMessageShouldReturnResult() throws Exception {
SendMessageDTO request = SendMessageDTO.builder()
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 8d1bdb8a4..cb5958992 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
@@ -88,7 +88,8 @@ class ToolInstanceRoutingTest {
MetadataService metadata = new MetadataService(globalMetadata,
globalAdmin, registry, targets,
mock(OperationAuditService.class), mock(MessageService.class),
mock(RuntimeAdminClientResolver.class),
-
mock(org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.class));
+
mock(org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.class),
+
mock(org.apache.rocketmq.studio.provider.LiteTopicProvider.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/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index 79476ca96..03a54214c 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
@@ -1147,6 +1147,110 @@ class RocketMQAdminClientImplTest {
assertThat(config.getAttributes()).containsEntry("+message.type",
TopicType.FIFO.name()));
}
+ @Test
+ void updateTopicShouldKeepLiteTopicExpirationOutOfTheWriteDeltaTest()
throws Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+ RmqTopic existing = new RmqTopic();
+ existing.setTopicType(TopicType.LITE.name());
+ existing.setWriteQueueNums(8);
+ existing.setReadQueueNums(8);
+ when(topicMapper.selectOne(any())).thenReturn(existing);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ // A LITE parent that already carries its TTL attribute: the generic
update path re-asserts
+ // message.type only, so the stored lite.topic.expiration must stay
out of the write delta.
+ TopicConfig stored = new TopicConfig("orders");
+ stored.setAttributes(Map.of("lite.topic.expiration", "1440"));
+ when(adminExt.examineTopicConfig(anyString(),
eq("orders"))).thenReturn(stored);
+ doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(),
any(TopicConfig.class));
+
+ TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
+ topic.setName("orders");
+ topic.setType(TopicType.LITE);
+ topic.setWriteQueues(16);
+
+ adminClient.updateTopic(topic);
+
+ ArgumentCaptor<TopicConfig> captor =
ArgumentCaptor.forClass(TopicConfig.class);
+ verify(adminExt).createAndUpdateTopicConfig(anyString(),
captor.capture());
+ assertThat(captor.getValue().getAttributes())
+ .containsEntry("+message.type", TopicType.LITE.name())
+ .containsOnlyKeys("+message.type");
+ }
+
+ @Test
+ void createTopicShouldWriteLiteTtlAttributeWithMessageTypeTest() throws
Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ when(topicMapper.selectOne(any())).thenReturn(null);
+ doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(),
any(TopicConfig.class));
+
+ TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
+ topic.setName("chat");
+ topic.setType(TopicType.LITE);
+ topic.setLiteTtlMinutes(1440);
+
+ adminClient.createTopic(topic);
+
+ ArgumentCaptor<TopicConfig> captor =
ArgumentCaptor.forClass(TopicConfig.class);
+ verify(adminExt).createAndUpdateTopicConfig(anyString(),
captor.capture());
+ assertThat(captor.getValue().getAttributes())
+ .containsEntry("+message.type", TopicType.LITE.name())
+ .containsEntry("+lite.topic.expiration", "1440");
+ }
+
+ @Test
+ void createTopicShouldIgnoreTtlMinutesForNonLiteTypesTest() throws
Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ when(topicMapper.selectOne(any())).thenReturn(null);
+ doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(),
any(TopicConfig.class));
+
+ TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
+ topic.setName("orders");
+ topic.setType(TopicType.NORMAL);
+ topic.setLiteTtlMinutes(1440);
+
+ adminClient.createTopic(topic);
+
+ ArgumentCaptor<TopicConfig> captor =
ArgumentCaptor.forClass(TopicConfig.class);
+ verify(adminExt).createAndUpdateTopicConfig(anyString(),
captor.capture());
+ assertThat(captor.getValue().getAttributes())
+ .containsEntry("+message.type", TopicType.NORMAL.name())
+ .doesNotContainKey("+lite.topic.expiration");
+ }
+
+ @Test
+ void updateTopicShouldIgnoreLiteTtlMinutesTest() throws Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+ RmqTopic existing = new RmqTopic();
+ existing.setTopicType(TopicType.LITE.name());
+ when(topicMapper.selectOne(any())).thenReturn(existing);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ TopicConfig stored = new TopicConfig("orders");
+ stored.setAttributes(Map.of("lite.topic.expiration", "1440"));
+ when(adminExt.examineTopicConfig(anyString(),
eq("orders"))).thenReturn(stored);
+ doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(),
any(TopicConfig.class));
+
+ // A stale or malicious update payload carries a TTL: the generic
update path must not
+ // write it — TTL is create-only and lives on the LiteTopic extension
endpoint.
+ TopicVO topic = new TopicVO();
+ topic.setInstanceId("instance-a");
+ topic.setName("orders");
+ topic.setType(TopicType.LITE);
+ topic.setLiteTtlMinutes(60);
+
+ adminClient.updateTopic(topic);
+
+ ArgumentCaptor<TopicConfig> captor =
ArgumentCaptor.forClass(TopicConfig.class);
+ verify(adminExt).createAndUpdateTopicConfig(anyString(),
captor.capture());
+ assertThat(captor.getValue().getAttributes())
+ .containsOnlyKeys("+message.type")
+ .containsEntry("+message.type", TopicType.LITE.name());
+ }
+
@Test
void updateTopicWithoutTypePreservesExistingType() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
@@ -2331,7 +2435,8 @@ class RocketMQAdminClientImplTest {
lenient().when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
return new MetadataService(mock(MetadataProvider.class), adminClient,
registry,
mock(InstanceResolver.class),
mock(OperationAuditService.class), mock(MessageService.class),
- runtimeAdminClientResolver, ownershipGuard);
+ runtimeAdminClientResolver, ownershipGuard,
+
mock(org.apache.rocketmq.studio.provider.LiteTopicProvider.class));
}
@Test
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index bfa08b729..02361644c 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -19,6 +19,8 @@ export interface Topic {
tps: number;
consumerGroupCount: number;
remark: string;
+ /** Create-only TTL of a Lite parent topic in minutes (type=LITE); never
sent on updates. */
+ liteTtlMinutes?: number;
gmtCreate: string;
gmtModified: string;
}
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index 3ef910a1e..b00b05ac9 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -3099,6 +3099,18 @@ const translations: Record<string, Record<Lang, string>>
= {
zh: '每个 Broker 节点写 {write} / 读 {read} 个队列；读写队列数一般情况下需保持一致',
en: '{write} write / {read} read queues per Broker node; keep them equal
in general',
},
+ 'topic.liteTtl': { zh: 'TTL（分钟）', en: 'TTL (minutes)' },
+ 'topic.liteTtlUnit': { zh: '分钟', en: 'min' },
+ 'topic.liteTtlRequired': { zh: '请输入 TTL', en: 'Please enter the TTL' },
+ 'topic.liteTtlExtra': {
+ zh:
+ '子 LiteTopic 超过 TTL 未发送新消息即自动过期回收；范围 1-43200（30 天）。创建后在 LiteTopic ' +
+ '页「延长 TTL」调整，那里以毫秒为单位（60000 毫秒 = 1 分钟）',
+ en:
+ 'Lite topics under this parent expire and are reclaimed after this many
minutes without new ' +
+ 'messages; range 1-43200 (30 days). Adjust it later from the LiteTopic
page, where Extend TTL ' +
+ 'takes milliseconds (60000 ms = 1 minute)',
+ },
'topic.remarkPlaceholder': {
zh: '可选，描述 Topic 用途',
en: 'Optional, describe the topic purpose',
@@ -3970,6 +3982,7 @@ const translations: Record<string, Record<Lang, string>>
= {
'liteTopic.status': { zh: '状态', en: 'Status' },
'liteTopic.actions': { zh: '操作', en: 'Actions' },
'liteTopic.extendTtl': { zh: '延长 TTL', en: 'Extend TTL' },
+ 'liteTopic.createParent': { zh: '创建 Lite 父主题', en: 'Create Lite parent' },
'liteTopic.viewSessions': { zh: '查看会话', en: 'View Sessions' },
'liteTopic.sessionDetail': { zh: '会话详情', en: 'Session Detail' },
'liteTopic.sessionId': { zh: '会话 ID', en: 'Session ID' },
diff --git a/web/src/pages/instance/__tests__/TopicPage.test.tsx
b/web/src/pages/instance/__tests__/TopicPage.test.tsx
index 3d3f9febd..daf494025 100644
--- a/web/src/pages/instance/__tests__/TopicPage.test.tsx
+++ b/web/src/pages/instance/__tests__/TopicPage.test.tsx
@@ -125,7 +125,9 @@ const healthyBrokerRoute: BrokerRoute = {
replicaCount: 1,
};
-const renderWithProviders = (initialEntry = '/instance/topic') =>
+const renderWithProviders = (
+ initialEntry: string | { pathname: string; state?: unknown } =
'/instance/topic',
+) =>
render(
<App>
<LangProvider>
@@ -376,6 +378,81 @@ describe('TopicPage', () => {
expect(within(getTableBody()).queryByText('create
response')).not.toBeInTheDocument();
});
+ it('sends the Lite TTL when creating a Lite parent topic', async () => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ renderWithProviders('/instance/instance-proxy-1/topic');
+
+ await screen.findByText('topic-01');
+ await user.click(screen.getByRole('button', { name: /创建 Topic/ }));
+ const dialog = await screen.findByRole('dialog', { name: /创建 Topic/ });
+ await user.type(within(dialog).getByLabelText('Topic 名称'), 'chat');
+ await user.click(within(dialog).getByText('LiteTopic'));
+
+ // The TTL input appears only for LITE creates and is prefilled with one
day (1440 minutes).
+ const ttl = within(dialog).getByLabelText('TTL（分钟）') as HTMLInputElement;
+ expect(ttl.value).toBe('1440');
+
+ await user.click(within(dialog).getByRole('button', { name: /创\s*建/ }));
+
+ await waitFor(() =>
expect(topicServiceMocks.createTopic).toHaveBeenCalledTimes(1));
+ expect(topicServiceMocks.createTopic).toHaveBeenCalledWith(
+ expect.objectContaining({ name: 'chat', type: 'LITE', liteTtlMinutes:
1440 }),
+ );
+ });
+
+ it('leaves the Lite TTL out of non-Lite creates', async () => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ renderWithProviders('/instance/instance-proxy-1/topic');
+
+ await screen.findByText('topic-01');
+ await user.click(screen.getByRole('button', { name: /创建 Topic/ }));
+ const dialog = await screen.findByRole('dialog', { name: /创建 Topic/ });
+ expect(within(dialog).queryByLabelText('TTL（分钟）')).not.toBeInTheDocument();
+ await user.type(within(dialog).getByLabelText('Topic 名称'), 'orders');
+ await user.click(within(dialog).getByRole('button', { name: /创\s*建/ }));
+
+ await waitFor(() =>
expect(topicServiceMocks.createTopic).toHaveBeenCalledTimes(1));
+
expect(topicServiceMocks.createTopic.mock.calls[0][0]).not.toHaveProperty('liteTtlMinutes');
+ });
+
+ it('never sends the Lite TTL on topic updates', async () => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ const liteTopic = { ...buildTopics(1)[0], type: 'LITE' };
+ mockTopicsList([liteTopic]);
+ renderWithProviders('/instance/instance-proxy-1/topic');
+
+ await screen.findByText('topic-01');
+ const row = screen.getByText('topic-01').closest('tr') as HTMLElement;
+ await user.click(within(row).getByRole('button', { name: /配置/ }));
+ const dialog = await screen.findByRole('dialog', { name: /编辑 Topic/ });
+
+ // TTL is create-only: the edit dialog has no TTL input even for a Lite
parent topic.
+ expect(within(dialog).queryByLabelText('TTL（分钟）')).not.toBeInTheDocument();
+
+ await user.click(within(dialog).getByRole('button', { name: /保\s*存/ }));
+
+ await waitFor(() =>
expect(topicServiceMocks.updateTopic).toHaveBeenCalledTimes(1));
+
expect(topicServiceMocks.updateTopic.mock.calls[0][0]).not.toHaveProperty('liteTtlMinutes');
+ });
+
+ it('opens the create dialog with LITE preselected for the LiteTopic page
action', async () => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ renderWithProviders({
+ pathname: '/instance/instance-proxy-1/topic',
+ state: { createType: 'LITE' },
+ });
+
+ const dialog = await screen.findByRole('dialog', { name: /创建 Topic/ });
+ expect(within(dialog).getByLabelText('TTL（分钟）')).toBeInTheDocument();
+ await user.type(within(dialog).getByLabelText('Topic 名称'), 'chat');
+ await user.click(within(dialog).getByRole('button', { name: /创\s*建/ }));
+
+ await waitFor(() =>
expect(topicServiceMocks.createTopic).toHaveBeenCalledTimes(1));
+ expect(topicServiceMocks.createTopic).toHaveBeenCalledWith(
+ expect.objectContaining({ name: 'chat', type: 'LITE', liteTtlMinutes:
1440 }),
+ );
+ });
+
it('updates topic config through the update service and reloads the page',
async () => {
topicServiceMocks.updateTopic.mockResolvedValue({
...buildTopics(1)[0],
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index ad75998a3..712bd124f 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -16,7 +16,7 @@
*/
import { useCallback, useEffect, useState, useMemo, useRef } from 'react';
-import { useNavigate } from 'react-router-dom';
+import { useLocation, useNavigate } from 'react-router-dom';
import {
Alert,
Table,
@@ -294,6 +294,7 @@ const TopicPageContent = ({
}: TopicPageContentProps) => {
const { t } = useLang();
const navigate = useNavigate();
+ const location = useLocation();
const isCloudInstance =
selectedInstance?.vendor === 'ALIYUN' || selectedInstance?.vendor ===
'TENCENT';
const canSendTestMessage = (topic: Topic) => !isCloudInstance || topic.type
=== 'NORMAL';
@@ -338,6 +339,21 @@ const TopicPageContent = ({
// 队列数提示要跟着输入值走（用户约定 2026-10-16：写死的「8 个队列」不随值变化是错的）
const createWriteQueues = Form.useWatch('writeQueues', form);
const createReadQueues = Form.useWatch('readQueues', form);
+
+ // The LiteTopic page's "Create Lite parent" action lands here with LITE
preselected and
+ // opens this same create dialog; the state is cleared so refresh/back
cannot reopen it.
+ // The handoff is consumed on a microtask (same shape as useDraftHandoff) so
the dialog opens
+ // outside the effect pass instead of stacking renders on it.
+ useEffect(() => {
+ const state = location.state as { createType?: string } | null;
+ if (state?.createType !== 'LITE') return;
+ void Promise.resolve().then(() => {
+ form.setFieldsValue({ type: 'LITE', liteTtlMinutes: 1440 });
+ setEditingTopic(null);
+ setModalOpen(true);
+ navigate('.', { replace: true, state: {} });
+ });
+ }, [location.state, form, navigate]);
const [sendModalOpen, setSendModalOpen] = useState(false);
const [sendTopic, setSendTopic] = useState<Topic | null>(null);
const [sending, setSending] = useState(false);
@@ -1001,17 +1017,20 @@ const TopicPageContent = ({
setCreating(true);
try {
const values = await form.validateFields();
+ // The Lite TTL is create-only: updates never carry it and non-Lite
creates leave it out.
+ const { liteTtlMinutes, ...rest } = values;
if (editingTopic) {
const updated = await updateTopic({
- ...values,
+ ...rest,
instanceId: selectedInstanceId,
});
await reloadTopicPage();
message.success(`Topic「${updated.name}」更新成功`);
} else {
const created = await createTopic({
- ...values,
+ ...rest,
instanceId: selectedInstanceId,
+ ...(values.type === 'LITE' ? { liteTtlMinutes } : {}),
});
await reloadTopicPage();
message.success(`Topic「${created.name}」创建成功`);
@@ -1523,6 +1542,7 @@ const TopicPageContent = ({
readQueues: 8,
perm: 'RW',
type: 'NORMAL',
+ liteTtlMinutes: 1440,
}}
style={{ marginTop: 16 }}
>
@@ -1567,6 +1587,23 @@ const TopicPageContent = ({
/>
</Form.Item>
+ {/* A Lite parent topic's TTL is written at create time and never
edited afterwards. */}
+ {createTopicType === 'LITE' && !editingTopic && (
+ <Form.Item
+ label={t('topic.liteTtl')}
+ name="liteTtlMinutes"
+ rules={[{ required: true, message: t('topic.liteTtlRequired') }]}
+ extra={t('topic.liteTtlExtra')}
+ >
+ <InputNumber
+ min={1}
+ max={43200}
+ style={{ width: '100%' }}
+ addonAfter={t('topic.liteTtlUnit')}
+ />
+ </Form.Item>
+ )}
+
{!isCloudInstance && (
<>
<Row gutter={16}>
diff --git a/web/src/pages/studio/LiteTopic.tsx
b/web/src/pages/studio/LiteTopic.tsx
index be301bb67..67ddd66b4 100644
--- a/web/src/pages/studio/LiteTopic.tsx
+++ b/web/src/pages/studio/LiteTopic.tsx
@@ -46,7 +46,9 @@ import {
Gauge,
Info,
DownloadSimple,
+ Plus,
} from '@phosphor-icons/react';
+import { useNavigate } from 'react-router-dom';
import PageHeader from '../../components/PageHeader';
import InfoBanner from '../../components/InfoBanner';
import { useLang } from '../../i18n/LangContext';
@@ -140,6 +142,7 @@ const collectNamespaces = (items: LiteTopicItem[]):
string[] => {
const LiteTopicPage: React.FC = () => {
const { t } = useLang();
const { message } = App.useApp();
+ const navigate = useNavigate();
const messageRef = useRef(message);
const translationRef = useRef(t);
@@ -876,6 +879,14 @@ const LiteTopicPage: React.FC = () => {
title={t('liteTopic.title')}
extra={
<Space>
+ <Button
+ type="primary"
+ icon={<Plus size={14} />}
+ size="small"
+ onClick={() => navigate('/instance/topic', { state: {
createType: 'LITE' } })}
+ >
+ {t('liteTopic.createParent')}
+ </Button>
<Button
icon={<DownloadSimple size={14} />}
size="small"
diff --git a/web/src/pages/studio/__tests__/LiteTopic.test.tsx
b/web/src/pages/studio/__tests__/LiteTopic.test.tsx
index e645bcfa8..77d4cf05b 100644
--- a/web/src/pages/studio/__tests__/LiteTopic.test.tsx
+++ b/web/src/pages/studio/__tests__/LiteTopic.test.tsx
@@ -18,6 +18,7 @@
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { act, render, screen, waitFor } from '@testing-library/react';
import userEvent from '@testing-library/user-event';
+import { MemoryRouter, Route, Routes } from 'react-router-dom';
import { App } from 'antd';
import { LangProvider } from '../../../i18n/LangContext';
import type { LiteTopicItem, LiteTopicQuota } from '../../../api/liteTopic';
@@ -75,7 +76,12 @@ const renderPage = () =>
render(
<App>
<LangProvider>
- <LiteTopic />
+ <MemoryRouter initialEntries={['/studio/lite-topic']}>
+ <Routes>
+ <Route path="/studio/lite-topic" element={<LiteTopic />} />
+ <Route path="/instance/topic"
element={<div>topic-create-page</div>} />
+ </Routes>
+ </MemoryRouter>
</LangProvider>
</App>,
);
@@ -358,6 +364,16 @@ describe('LiteTopic Page', () => {
);
});
+ it('routes to the generic topic page to create a Lite parent', async () => {
+ const user = userEvent.setup();
+ renderPage();
+
+ await screen.findByText('order-*');
+ await user.click(screen.getByRole('button', { name: /创建 Lite 父主题/ }));
+
+ expect(await screen.findByText('topic-create-page')).toBeInTheDocument();
+ });
+
it('returns to the first page when the local TTL filter changes', async ()
=> {
apiMocks.queryLiteTopicList.mockResolvedValue([
...Array.from({ length: 11 }, (_, index) => ({
