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 e5ac52440 feat(studio): serve LiteTopic from the broker lite admin API
(#4220)
e5ac52440 is described below
commit e5ac52440668edf9a9c00af95e53777f19765dfe
Author: zhaohai <[email protected]>
AuthorDate: Wed Sep 16 16:21:05 2026 +0800
feat(studio): serve LiteTopic from the broker lite admin API (#4220)
* feat(studio): serve LiteTopic from the broker lite admin API
The LiteTopic console page was fully built but every endpoint answered 501
because LiteTopicService was a stub, so the feature only existed on screen.
Wire the service to the RocketMQ lite admin RPCs, all of which are reachable
through the already declared rocketmq-tools 5.5.0 dependency:
GET_BROKER_LITE_INFO -> parent topics, TTL attribute, consumer groups
GET_PARENT_TOPIC_INFO -> per-parent lite topic count and TTL
GET_LITE_CLIENT_INFO -> a session's lite topic set and last access time
GET_LITE_GROUP_INFO -> pending lag and consumed offsets
Parent topics are identified with LiteUtil, the same helper the broker uses,
instead of guessing name prefixes. A session is keyed by parent topic, group
and client id, and is resolved against the broker master that actually owns
that client's lite subscription.
Everything reported comes from the broker; values the broker does not expose
are left null rather than fabricated. TTLs cross the API in milliseconds
(the
console contract) and are stored in minutes (the protocol unit). Scans are
bounded per request so a page load stays predictable.
The controller contract and the existing web UI are unchanged.
* fix(studio): alter the lite TTL with the broker's +key attribute protocol
Attributes read back from examineTopicConfig use bare keys, but the broker
validates updates through AttributeParser, which only accepts the +key=value
change form. Re-sending bare message.type made extendTTL fail against a real
broker with 'add/alter attribute format is wrong: message.type'.
extendTTL now sends only the +lite.topic.expiration change entry; the broker
merges that entry into the stored attributes, so message.type and everything
else stays untouched. Verified against a live RocketMQ 5.5.1 broker: the
change is accepted and the new TTL is served back by the lite admin API.
* fix(studio): build the lite quota ratio from one consistent master set
---
.../studio/instance/topic/LiteTopicService.java | 120 +++-
.../rocketmq/studio/model/LiteTopicSummary.java | 2 +
.../studio/provider/LiteTopicProvider.java | 86 +++
.../provider/apache/RocketMQLiteTopicProvider.java | 612 +++++++++++++++++++++
.../instance/topic/LiteTopicServiceTest.java | 152 ++++-
.../apache/RocketMQLiteTopicProviderTest.java | 386 +++++++++++++
6 files changed, 1335 insertions(+), 23 deletions(-)
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 6d70fb4cb..caf9cf7f0 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
@@ -18,21 +18,42 @@
package org.apache.rocketmq.studio.instance.topic;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.model.LiteTopicQuota;
+import org.apache.rocketmq.studio.model.LiteTopicSession;
+import org.apache.rocketmq.studio.model.LiteTopicSummary;
+import org.apache.rocketmq.studio.provider.LiteTopicProvider;
+import org.springframework.util.StringUtils;
+import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
+import java.util.Date;
import java.util.List;
+/**
+ * Read/write facade for the LiteTopic console page.
+ *
+ * <p>All broker interaction is delegated to the {@link LiteTopicProvider};
this class validates
+ * request input and maps the provider's domain models onto the REST view
objects the console
+ * consumes. TTLs are exchanged with the console in milliseconds (matching the
UI contract) while
+ * the broker stores them in minutes — the provider performs that conversion.
+ */
@Service
+@RequiredArgsConstructor
public class LiteTopicService {
- private static final String PROVIDER_UNAVAILABLE_MESSAGE = "LiteTopic
provider integration is not available";
- private static final int NOT_IMPLEMENTED = 501;
+
+ private final LiteTopicProvider liteTopicProvider;
public List<LiteTopicItemVO> listLiteTopics(String pattern, String
namespace) {
- throw new BusinessException(NOT_IMPLEMENTED,
PROVIDER_UNAVAILABLE_MESSAGE);
+ return liteTopicProvider.listLiteTopics(pattern, namespace).stream()
+ .map(this::toItemVO)
+ .toList();
}
public LiteTopicSessionVO getSession(String sessionId) {
- throw new BusinessException(NOT_IMPLEMENTED,
PROVIDER_UNAVAILABLE_MESSAGE);
+ if (!StringUtils.hasText(sessionId)) {
+ throw new BusinessException(400, "sessionId is required");
+ }
+ return toSessionVO(liteTopicProvider.getSession(sessionId));
}
public void extendTTL(String topicPattern, Long newTTL) {
@@ -42,14 +63,99 @@ public class LiteTopicService {
if (newTTL == null || newTTL <= 0) {
throw new BusinessException(400, "newTTL must be positive");
}
- throw new BusinessException(NOT_IMPLEMENTED,
PROVIDER_UNAVAILABLE_MESSAGE);
+ liteTopicProvider.extendTTL(topicPattern.trim(), newTTL);
}
public LiteTopicQuotaVO getQuota(String namespace) {
- throw new BusinessException(NOT_IMPLEMENTED,
PROVIDER_UNAVAILABLE_MESSAGE);
+ return toQuotaVO(liteTopicProvider.getQuota(namespace));
}
public LiteTopicCapabilityVO getCapability() {
- return new LiteTopicCapabilityVO(false);
+ return new LiteTopicCapabilityVO(liteTopicProvider.isSupported());
+ }
+
+ // ─── Mapping ──────────────────────────────────────────────────────
+
+ private LiteTopicItemVO toItemVO(LiteTopicSummary summary) {
+ return LiteTopicItemVO.builder()
+ .topicPattern(summary.getTopicPattern())
+ .namespace(summary.getNamespace())
+ .topicCount(summary.getTopicCount())
+ .consumerCount(summary.getConsumerCount())
+ .totalBacklog(summary.getTotalBacklog())
+ .averageTTL(summary.getAverageTTL())
+ .ttlStatus(summary.getTTLStatus())
+ .lastActiveTime(epochMillis(summary.getLastActiveTime()))
+ .sessionIds(summary.getSessionIds())
+ .build();
+ }
+
+ private LiteTopicSessionVO toSessionVO(LiteTopicSession session) {
+ return LiteTopicSessionVO.builder()
+ .sessionId(session.getSessionId())
+ .clientId(session.getClientId())
+ .clientAddress(session.getClientAddress())
+ .parentTopic(session.getParentTopic())
+ .consumerGroup(session.getConsumerGroup())
+ .createTime(epochMillis(session.getCreateTime()))
+ .lastActiveTime(epochMillis(session.getLastActiveTime()))
+ .ttl(session.getTtl())
+ .ttlRemaining(session.getTtlRemaining())
+ .status(session.getStatus())
+ .totalMessages(session.getTotalMessages())
+ .consumedMessages(session.getConsumedMessages())
+ .pendingMessages(session.getPendingMessages())
+ .liteTopicCreationCount(session.getLiteTopicCreationCount())
+ .liteTopics(toLiteTopicRows(session))
+ .build();
+ }
+
+ /**
+ * A lite topic shares its parent topic's TTL policy, so every entry
inherits the session's
+ * status and remaining TTL.
+ */
+ private List<LiteTopicSessionVO.SessionLiteTopic>
toLiteTopicRows(LiteTopicSession session) {
+ if (session.getLiteTopics() == null) {
+ return List.of();
+ }
+ return session.getLiteTopics().stream()
+ .map(name -> new LiteTopicSessionVO.SessionLiteTopic(
+ name, session.getStatus(), session.getTtlRemaining()))
+ .toList();
+ }
+
+ private LiteTopicQuotaVO toQuotaVO(LiteTopicQuota quota) {
+ return LiteTopicQuotaVO.builder()
+ .currentTopicCount(quota.getCurrentTopicCount())
+ .maxTopicCount(quota.getMaxTopicCount())
+ .currentSessionCount(quota.getCurrentSessionCount())
+ .maxSessionCount(quota.getMaxSessionCount())
+ .currentCreationRate(toInt(quota.getCurrentCreationRate()))
+ .maxCreationRate(toInt(quota.getMaxCreationRate()))
+ .usageRate(quota.getUsageRate())
+ .sessionUsageRate(quota.getSessionUsageRate())
+ .defaultTTL(quota.getDefaultTTL())
+ .maxTTL(quota.getMaxTTL())
+ .remainingQuota(quota.getRemainingQuota())
+ .consumerDensity(consumerDensity(quota))
+ .build();
+ }
+
+ /** Sessions per lite topic, i.e. how many independent consumers share the
parent topic. */
+ private Double consumerDensity(LiteTopicQuota quota) {
+ Integer topics = quota.getCurrentTopicCount();
+ Integer sessions = quota.getCurrentSessionCount();
+ if (topics == null || topics <= 0 || sessions == null) {
+ return null;
+ }
+ return (double) sessions / topics;
+ }
+
+ private static Long epochMillis(Date value) {
+ return value == null ? null : value.getTime();
+ }
+
+ private static Integer toInt(Double value) {
+ return value == null ? null : value.intValue();
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/model/LiteTopicSummary.java
b/server/src/main/java/org/apache/rocketmq/studio/model/LiteTopicSummary.java
index 3470d3db0..1d5041eae 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/model/LiteTopicSummary.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/model/LiteTopicSummary.java
@@ -25,6 +25,8 @@ public class LiteTopicSummary {
private String topicPattern;
+ private String namespace;
+
private Integer topicCount;
private List<String> sessionIds;
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
new file mode 100644
index 000000000..7281cfd19
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/LiteTopicProvider.java
@@ -0,0 +1,86 @@
+/*
+ * 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;
+
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.model.LiteTopicQuota;
+import org.apache.rocketmq.studio.model.LiteTopicSession;
+import org.apache.rocketmq.studio.model.LiteTopicSummary;
+
+import java.util.List;
+
+/**
+ * LiteTopic (LMQ-backed light message queue) inspection SPI.
+ *
+ * <p>LiteTopic is a broker-side feature: a parent topic declared with
+ * {@code TopicMessageType.LITE} stores each "lite topic" in its own LMQ whose
name is
+ * {@code %LMQ%$parentTopic$liteTopic}. The studio queries the broker directly
through the
+ * RocketMQ lite admin RPCs ({@code GET_BROKER_LITE_INFO}, {@code
GET_PARENT_TOPIC_INFO},
+ * {@code GET_LITE_CLIENT_INFO}, {@code GET_LITE_GROUP_INFO}).
+ *
+ * <p>Implementations that cannot reach a lite-capable broker must report
+ * {@link #isSupported()} as {@code false} and leave the default methods
untouched, so the
+ * console degrades to an informational "not supported" state instead of
failing.
+ */
+public interface LiteTopicProvider {
+
+ String UNSUPPORTED = "LiteTopic is not supported by this provider";
+ int NOT_IMPLEMENTED = 501;
+
+ /**
+ * 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
+ * unsupported-code error.
+ */
+ default boolean isSupported() {
+ return false;
+ }
+
+ /**
+ * Aggregates every lite parent topic visible on the cluster into one
summary per parent topic.
+ *
+ * @param pattern optional substring filter applied to the parent topic
name
+ * @param namespace optional namespace filter; {@code null} or blank
returns every namespace
+ */
+ default List<LiteTopicSummary> listLiteTopics(String pattern, String
namespace) {
+ throw new BusinessException(NOT_IMPLEMENTED, UNSUPPORTED);
+ }
+
+ /**
+ * Resolves one "session" — a single client's lite topic set for one
parent topic and group.
+ * The session id is an opaque token produced by {@link #listLiteTopics}.
+ */
+ default LiteTopicSession getSession(String sessionId) {
+ throw new BusinessException(NOT_IMPLEMENTED, UNSUPPORTED);
+ }
+
+ /**
+ * Rewrites the parent topic's {@code lite.topic.expiration} attribute
(the single TTL policy
+ * shared by every lite topic under that parent).
+ *
+ * @param topicPattern the parent topic name
+ * @param ttlMillis the new TTL in milliseconds
+ */
+ default void extendTTL(String topicPattern, long ttlMillis) {
+ throw new BusinessException(NOT_IMPLEMENTED, UNSUPPORTED);
+ }
+
+ /** Cluster-wide LiteTopic quota, aggregated across broker masters. */
+ default LiteTopicQuota getQuota(String namespace) {
+ throw new BusinessException(NOT_IMPLEMENTED, UNSUPPORTED);
+ }
+}
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
new file mode 100644
index 000000000..f86f69a80
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
@@ -0,0 +1,612 @@
+/*
+ * 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 org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
+import org.apache.rocketmq.common.lite.LiteUtil;
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import
org.apache.rocketmq.remoting.protocol.body.GetBrokerLiteInfoResponseBody;
+import
org.apache.rocketmq.remoting.protocol.body.GetLiteClientInfoResponseBody;
+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.common.exception.BusinessException;
+import org.apache.rocketmq.studio.model.LiteTopicQuota;
+import org.apache.rocketmq.studio.model.LiteTopicSession;
+import org.apache.rocketmq.studio.model.LiteTopicSummary;
+import org.apache.rocketmq.studio.provider.LiteTopicProvider;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Live {@link LiteTopicProvider} backed by the RocketMQ broker lite admin
RPCs.
+ *
+ * <p>Everything reported here comes from the broker: parent topics (and their
TTL) from
+ * {@code GET_BROKER_LITE_INFO}, per-parent lite topic counts from {@code
GET_PARENT_TOPIC_INFO},
+ * sessions from the consumer connections plus {@code GET_LITE_CLIENT_INFO},
and backlog from
+ * {@code GET_LITE_GROUP_INFO}. No name-prefix guessing is involved — a lite
topic is identified
+ * through {@link LiteUtil}, which is the same helper the broker uses.
+ *
+ * <p>The console reaches lite endpoints without an instance id, so the
queries run against the
+ * default configured NameServer, exactly like the other legacy cluster-scoped
metadata calls.
+ *
+ * <p>Data the broker does not expose is left {@code null} rather than
fabricated: a session has no
+ * broker-side creation timestamp, and there is no per-parent "last active"
clock, so the newest
+ * client {@code lastAccessTime} under the parent is used as the closest real
signal.
+ */
+@Slf4j
+@Service
+@Primary
+@RequiredArgsConstructor
+public class RocketMQLiteTopicProvider implements LiteTopicProvider {
+
+ /** Namespace label used for broker-created resources, which carry no
namespace of their own. */
+ static final String DEFAULT_NAMESPACE = "DEFAULT";
+
+ /** Separator inside the opaque session id; no topic or consumer group
name may contain it. */
+ static final String SESSION_SEPARATOR = "~";
+
+ /** Upper bound on parent topics resolved per list call, to keep one page
load bounded. */
+ static final int MAX_LITE_TOPIC_SCAN = 200;
+
+ /** Upper bound on sessions resolved per parent topic. */
+ static final int MAX_LITE_SESSION_SCAN = 500;
+
+ /** 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;
+
+ // ─── Capability ───────────────────────────────────────────────────
+
+ @Override
+ public boolean isSupported() {
+ if (!hasAdmin()) {
+ return false;
+ }
+ try {
+ return
Boolean.TRUE.equals(adminFactory.execute(properties.getNamesrvAddr(), null,
admin -> {
+ List<String> masters = masterAddresses(admin);
+ return !masters.isEmpty() &&
admin.getBrokerLiteInfo(masters.get(0)) != null;
+ }));
+ } catch (Exception probeFailure) {
+ // An older broker answers the lite RPC with an unsupported-code
error; that is the
+ // signal the console degrades on, so report it as unsupported
instead of failing.
+ log.debug("LiteTopic capability probe failed: {}",
probeFailure.getMessage());
+ return false;
+ }
+ }
+
+ // ─── List ─────────────────────────────────────────────────────────
+
+ @Override
+ public List<LiteTopicSummary> listLiteTopics(String pattern, String
namespace) {
+ requireAdmin();
+ return execute(admin -> {
+ List<String> masters = masterAddresses(admin);
+ if (masters.isEmpty()) {
+ throw new BusinessException(503, "No broker master is
available for LiteTopic queries");
+ }
+ Map<String, ParentTopicAccumulator> parents =
discoverParentTopics(admin, masters);
+ List<LiteTopicSummary> summaries = new ArrayList<>();
+ int scanned = 0;
+ for (ParentTopicAccumulator parent : parents.values()) {
+ if (!matchesPattern(parent.parentTopic, pattern)) {
+ continue;
+ }
+ if (StringUtils.hasText(namespace)
+ &&
!DEFAULT_NAMESPACE.equalsIgnoreCase(namespace.trim())) {
+ continue;
+ }
+ if (scanned++ >= MAX_LITE_TOPIC_SCAN) {
+ log.warn("LiteTopic list truncated at {} parent topics",
MAX_LITE_TOPIC_SCAN);
+ break;
+ }
+ summaries.add(buildSummary(admin, masters, parent));
+ }
+
summaries.sort(Comparator.comparing(LiteTopicSummary::getTopicPattern,
+ Comparator.nullsLast(String::compareTo)));
+ return summaries;
+ });
+ }
+
+ private Map<String, ParentTopicAccumulator>
discoverParentTopics(MQAdminExt admin, List<String> masters)
+ throws Exception {
+ Map<String, ParentTopicAccumulator> parents = new LinkedHashMap<>();
+ for (String master : masters) {
+ GetBrokerLiteInfoResponseBody info =
admin.getBrokerLiteInfo(master);
+ if (info == null || info.getTopicMeta() == null) {
+ continue;
+ }
+ info.getTopicMeta().forEach((parent, ttlMinutes) -> parents
+ .computeIfAbsent(parent, key -> new
ParentTopicAccumulator(key, master))
+ .merge(master, ttlMinutes));
+ if (info.getGroupMeta() != null) {
+ info.getGroupMeta().forEach((parent, groups) -> {
+ if (groups == null || groups.isEmpty()) {
+ return;
+ }
+ // A parent topic can be sharded across brokers; the group
binding is reported
+ // by whichever broker owns the lite topic, so union
across masters.
+ parents.computeIfAbsent(parent, key -> new
ParentTopicAccumulator(key, master))
+ .groups.addAll(groups);
+ });
+ }
+ }
+ return parents;
+ }
+
+ private LiteTopicSummary buildSummary(MQAdminExt admin, List<String>
masters, ParentTopicAccumulator parent) {
+ LiteTopicSummary summary = new LiteTopicSummary();
+ summary.setTopicPattern(parent.parentTopic);
+ summary.setNamespace(DEFAULT_NAMESPACE);
+
+ Long ttlMillis = toMillis(parent.ttlMinutes);
+ summary.setAverageTTL(ttlMillis);
+ summary.setMinTTL(ttlMillis);
+ summary.setMaxTTL(ttlMillis);
+
+ int topicCount = 0;
+ try {
+ GetParentTopicInfoResponseBody parentInfo =
admin.getParentTopicInfo(parent.brokerAddr, parent.parentTopic);
+ if (parentInfo != null) {
+ topicCount = Math.max(parentInfo.getLiteTopicCount(), 0);
+ }
+ } catch (Exception failure) {
+ log.debug("Failed to read parent topic info for {}: {}",
parent.parentTopic, failure.getMessage());
+ }
+
+ long totalBacklog = 0;
+ Long lastActive = null;
+ Set<String> sessionIds = new LinkedHashSet<>();
+ int consumerCount = 0;
+ int sessionBudget = MAX_LITE_SESSION_SCAN;
+ for (String group : parent.groups) {
+ totalBacklog += groupLag(admin, parent.brokerAddr, group);
+ for (Connection connection : consumerConnections(admin, group)) {
+ if (sessionBudget-- <= 0) {
+ log.warn("LiteTopic session scan for {} truncated at {}
sessions",
+ parent.parentTopic, MAX_LITE_SESSION_SCAN);
+ break;
+ }
+ consumerCount++;
+ sessionIds.add(encodeSessionId(parent.parentTopic, group,
connection.getClientId()));
+ Long active = clientLastAccess(admin, masters,
parent.parentTopic, group,
+ connection.getClientId());
+ if (active != null && (lastActive == null || active >
lastActive)) {
+ lastActive = active;
+ }
+ }
+ }
+
+ summary.setTopicCount(topicCount);
+ summary.setConsumerCount(consumerCount);
+ summary.setTotalBacklog(totalBacklog);
+ summary.setSessionIds(new ArrayList<>(sessionIds));
+ summary.setActive(consumerCount > 0);
+ if (lastActive != null) {
+ summary.setLastActiveTime(new Date(lastActive));
+ }
+ return summary;
+ }
+
+ // ─── Session detail ───────────────────────────────────────────────
+
+ @Override
+ public LiteTopicSession getSession(String sessionId) {
+ requireAdmin();
+ String[] parts = decodeSessionId(sessionId);
+ String parentTopic = parts[0];
+ String group = parts[1];
+ String clientId = parts[2];
+ return execute(admin -> {
+ List<String> masters = masterAddresses(admin);
+ LocatedClient located = locateClient(admin, masters, parentTopic,
group, clientId);
+ if (located == null) {
+ throw new BusinessException(404, "LiteTopic session not found:
" + sessionId);
+ }
+ return buildSession(admin, located, sessionId, parentTopic, group,
clientId);
+ });
+ }
+
+ private LiteTopicSession buildSession(MQAdminExt admin, LocatedClient
located, String sessionId,
+ String parentTopic, String group,
String clientId) {
+ GetLiteClientInfoResponseBody clientInfo = located.body;
+ LiteTopicSession session = new LiteTopicSession();
+ session.setSessionId(sessionId);
+ session.setClientId(clientId);
+ session.setClientAddress(clientAddress(admin, group, clientId));
+ session.setParentTopic(parentTopic);
+ session.setConsumerGroup(group);
+ session.setLiteTopicCreationCount(clientInfo.getLiteTopicCount() >= 0
+ ? clientInfo.getLiteTopicCount() : null);
+
+ Long ttlMillis = toMillis(parentTtlMinutes(admin, located.master,
parentTopic));
+ session.setTtl(ttlMillis);
+
+ Set<String> lmqSet = clientInfo.getLiteTopicSet() == null
+ ? Set.of() : clientInfo.getLiteTopicSet();
+ List<String> liteTopics = lmqSet.stream()
+ .map(LiteUtil::getLiteTopic)
+ .filter(Objects::nonNull)
+ .sorted()
+ .toList();
+ session.setLiteTopics(new LinkedHashSet<>(liteTopics));
+
+ long pending = groupLag(admin, located.master, group);
+ long consumed = consumedMessages(admin, located.master, group,
liteTopics);
+ session.setPendingMessages(pending);
+ session.setConsumedMessages(consumed);
+ session.setTotalMessages(consumed + pending);
+ session.setConsumptionRate(session.getConsumptionProgress());
+
+ long lastAccess = clientInfo.getLastAccessTime();
+ if (lastAccess > 0) {
+ session.setLastActiveTime(new Date(lastAccess));
+ }
+ applyTtlState(session, ttlMillis, lastAccess > 0 ? lastAccess : null);
+ return session;
+ }
+
+ private void applyTtlState(LiteTopicSession session, Long ttlMillis, Long
lastAccess) {
+ if (ttlMillis == null || lastAccess == null) {
+ // No expiration attribute means the broker never expires this
session's lite topics.
+ session.setStatus("ACTIVE");
+ return;
+ }
+ long remaining = ttlMillis - (System.currentTimeMillis() - lastAccess);
+ session.setTtlRemaining(Math.max(remaining, 0));
+ session.setStatus(remaining > 0 ? "ACTIVE" : "EXPIRED");
+ }
+
+ private long consumedMessages(MQAdminExt admin, String brokerAddr, String
group, List<String> liteTopics) {
+ long consumed = 0;
+ int scanned = 0;
+ for (String liteTopic : liteTopics) {
+ if (scanned++ >= MAX_SESSION_LITE_TOPIC_SCAN) {
+ log.warn("LiteTopic session consumed-offset scan truncated at
{} lite topics",
+ MAX_SESSION_LITE_TOPIC_SCAN);
+ break;
+ }
+ try {
+ GetLiteGroupInfoResponseBody body =
admin.getLiteGroupInfo(brokerAddr, group, liteTopic, 1);
+ OffsetWrapper wrapper = body == null ? null :
body.getLiteTopicOffsetWrapper();
+ if (wrapper != null && wrapper.getConsumerOffset() > 0) {
+ consumed += wrapper.getConsumerOffset();
+ }
+ } catch (Exception failure) {
+ log.debug("Failed to read lite offset for {}|{}: {}", group,
liteTopic, failure.getMessage());
+ }
+ }
+ return consumed;
+ }
+
+ // ─── TTL update ───────────────────────────────────────────────────
+
+ @Override
+ public void extendTTL(String topicPattern, long ttlMillis) {
+ requireAdmin();
+ if (!StringUtils.hasText(topicPattern)) {
+ throw new BusinessException(400, "topicPattern is required");
+ }
+ 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 -> {
+ int updated = 0;
+ for (String master : masterAddresses(admin)) {
+ TopicConfig config = liteTopicConfig(admin, master,
topicPattern);
+ if (config == null) {
+ continue;
+ }
+ // Attributes read back from the broker use bare keys
("lite.topic.expiration"),
+ // while the update protocol only accepts change entries
("+key=value"); a bare
+ // key is rejected with "add/alter attribute format is wrong".
The broker merges
+ // this change set into the stored attributes, so re-sending
message.type is
+ // unnecessary - and message.type is validated as immutable on
alter anyway.
+ // Only the TTL is altered.
+ Map<String, String> change = new HashMap<>();
+ change.put("+lite.topic.expiration", String.valueOf(minutes));
+ config.setAttributes(change);
+ admin.createAndUpdateTopicConfig(master, config);
+ updated++;
+ }
+ if (updated == 0) {
+ throw new BusinessException(404, "Lite parent topic not found:
" + topicPattern);
+ }
+ log.info("Extended LiteTopic TTL to {}ms ({} min) for parent topic
{} on {} broker(s)",
+ ttlMillis, minutes, topicPattern, updated);
+ return null;
+ });
+ }
+
+ private TopicConfig liteTopicConfig(MQAdminExt admin, String brokerAddr,
String topic) {
+ try {
+ TopicConfig config = admin.examineTopicConfig(brokerAddr, topic);
+ if (config == null ||
!TopicMessageType.LITE.equals(config.getTopicMessageType())) {
+ return null;
+ }
+ return config;
+ } catch (Exception failure) {
+ log.debug("Parent topic {} is not configured on {}: {}", topic,
brokerAddr, failure.getMessage());
+ return null;
+ }
+ }
+
+ // ─── Quota ────────────────────────────────────────────────────────
+
+ @Override
+ public LiteTopicQuota getQuota(String namespace) {
+ requireAdmin();
+ return execute(admin -> {
+ List<String> masters = masterAddresses(admin);
+ if (masters.isEmpty()) {
+ throw new BusinessException(503, "No broker master is
available for LiteTopic queries");
+ }
+ long currentTopics = 0;
+ long maxTopics = 0;
+ long currentSessions = 0;
+ long maxSessions = 0;
+ for (String master : masters) {
+ GetBrokerLiteInfoResponseBody info =
admin.getBrokerLiteInfo(master);
+ if (info == null) {
+ // Skip the master entirely: adding its session cap
without its current
+ // counts would build the ratio out of two different
master sets.
+ continue;
+ }
+ currentTopics += Math.max(info.getCurrentLmqNum(), 0);
+ maxTopics += Math.max(info.getMaxLmqNum(), 0);
+ currentSessions += Math.max(info.getLiteSubscriptionCount(),
0);
+ Properties brokerConfig = brokerConfig(admin, master);
+ if (brokerConfig != null) {
+ maxSessions +=
parsePositiveLong(brokerConfig.getProperty("maxLiteSubscriptionCount"));
+ }
+ }
+
+ LiteTopicQuota quota = new LiteTopicQuota();
+ quota.setCurrentTopicCount(toInt(currentTopics));
+ quota.setMaxTopicCount(toInt(maxTopics));
+ quota.setCurrentSessionCount(toInt(currentSessions));
+ quota.setMaxSessionCount(toInt(maxSessions));
+ // The protocol caps lite.topic.expiration at 30 days. There is no
namespace-level
+ // default TTL to report: the effective TTL comes from each parent
topic's own
+ // attribute, and the broker's minLiteTTl is a floor rather than a
default.
+ quota.setMaxTTL(TimeUnit.MINUTES.toMillis(MAX_LITE_TTL_MINUTES));
+ // No broker-side creation-rate quota exists; report zero so the
console renders a
+ // defined value instead of a blank gauge.
+ quota.setCurrentCreationRate(0.0);
+ quota.setMaxCreationRate(0.0);
+ return quota;
+ });
+ }
+
+ private Properties brokerConfig(MQAdminExt admin, String brokerAddr) {
+ try {
+ return admin.getBrokerConfig(brokerAddr);
+ } catch (Exception failure) {
+ log.debug("Failed to read broker config for {}: {}", brokerAddr,
failure.getMessage());
+ return null;
+ }
+ }
+
+ // ─── Admin plumbing ───────────────────────────────────────────────
+
+ private void requireAdmin() {
+ if (!hasAdmin()) {
+ throw new BusinessException(NOT_IMPLEMENTED, UNSUPPORTED);
+ }
+ }
+
+ private boolean hasAdmin() {
+ return StringUtils.hasText(properties.getNamesrvAddr());
+ }
+
+ private <T> T execute(MqAdminExtFactory.AdminAction<T> action) {
+ return adminFactory.execute(properties.getNamesrvAddr(), null, action);
+ }
+
+ private List<String> masterAddresses(MQAdminExt admin) throws Exception {
+ ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
+ if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+ return List.of();
+ }
+ List<String> masters = new ArrayList<>();
+ for (BrokerData brokerData :
clusterInfo.getBrokerAddrTable().values()) {
+ if (brokerData == null || brokerData.getBrokerAddrs() == null
+ || brokerData.getBrokerAddrs().isEmpty()) {
+ continue;
+ }
+ String master = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
+ if (master == null) {
+ master =
brokerData.getBrokerAddrs().values().iterator().next();
+ }
+ if (master != null) {
+ masters.add(master);
+ }
+ }
+ return masters;
+ }
+
+ /**
+ * A client's lite subscription lives on the broker that owns its channel,
so the session has to
+ * be resolved against the master that actually reports it (a negative
topic count means the
+ * broker does not hold that client's subscription).
+ */
+ private LocatedClient locateClient(MQAdminExt admin, List<String> masters,
String parentTopic,
+ String group, String clientId) {
+ for (String master : masters) {
+ try {
+ GetLiteClientInfoResponseBody body =
admin.getLiteClientInfo(master, parentTopic, group, clientId);
+ if (body != null && body.getLiteTopicCount() >= 0) {
+ return new LocatedClient(master, body);
+ }
+ } catch (Exception failure) {
+ log.debug("Lite client {} not resolvable on {}: {}", clientId,
master, failure.getMessage());
+ }
+ }
+ return null;
+ }
+
+ private Long clientLastAccess(MQAdminExt admin, List<String> masters,
String parentTopic,
+ String group, String clientId) {
+ LocatedClient located = locateClient(admin, masters, parentTopic,
group, clientId);
+ if (located == null || located.body.getLastAccessTime() <= 0) {
+ return null;
+ }
+ return located.body.getLastAccessTime();
+ }
+
+ private Set<Connection> consumerConnections(MQAdminExt admin, String
group) {
+ try {
+ ConsumerConnection connection =
admin.examineConsumerConnectionInfo(group);
+ if (connection == null || connection.getConnectionSet() == null) {
+ return Set.of();
+ }
+ return connection.getConnectionSet();
+ } catch (Exception failure) {
+ log.debug("Failed to read consumer connections for group {}: {}",
group, failure.getMessage());
+ return Set.of();
+ }
+ }
+
+ private String clientAddress(MQAdminExt admin, String group, String
clientId) {
+ for (Connection connection : consumerConnections(admin, group)) {
+ if (clientId != null && clientId.equals(connection.getClientId()))
{
+ return connection.getClientAddr();
+ }
+ }
+ return null;
+ }
+
+ private long groupLag(MQAdminExt admin, String brokerAddr, String group) {
+ try {
+ GetLiteGroupInfoResponseBody body =
admin.getLiteGroupInfo(brokerAddr, group, null, 1);
+ if (body == null) {
+ return 0;
+ }
+ return Math.max(body.getTotalLagCount(), 0);
+ } catch (Exception failure) {
+ log.debug("Failed to read lite backlog for group {} on {}: {}",
group, brokerAddr, failure.getMessage());
+ return 0;
+ }
+ }
+
+ private Integer parentTtlMinutes(MQAdminExt admin, String brokerAddr,
String parentTopic) {
+ try {
+ GetParentTopicInfoResponseBody body =
admin.getParentTopicInfo(brokerAddr, parentTopic);
+ return body == null ? null : body.getTtl();
+ } catch (Exception failure) {
+ log.debug("Failed to read TTL for parent topic {}: {}",
parentTopic, failure.getMessage());
+ return null;
+ }
+ }
+
+ // ─── Helpers ──────────────────────────────────────────────────────
+
+ static String encodeSessionId(String parentTopic, String group, String
clientId) {
+ return parentTopic + SESSION_SEPARATOR + group + SESSION_SEPARATOR +
clientId;
+ }
+
+ static String[] decodeSessionId(String sessionId) {
+ String[] parts = sessionId == null ? new String[0] :
sessionId.split(SESSION_SEPARATOR, 3);
+ if (parts.length != 3 || !StringUtils.hasText(parts[0])
+ || !StringUtils.hasText(parts[1]) ||
!StringUtils.hasText(parts[2])) {
+ throw new BusinessException(400, "Malformed LiteTopic session id");
+ }
+ return parts;
+ }
+
+ private static boolean matchesPattern(String value, String pattern) {
+ if (!StringUtils.hasText(pattern)) {
+ return true;
+ }
+ return value != null && value.toLowerCase(Locale.ROOT)
+ .contains(pattern.trim().toLowerCase(Locale.ROOT));
+ }
+
+ private static Long toMillis(Integer minutes) {
+ return minutes == null || minutes <= 0 ? null :
TimeUnit.MINUTES.toMillis(minutes);
+ }
+
+ private static long parsePositiveLong(String raw) {
+ if (raw == null || raw.isBlank()) {
+ return 0;
+ }
+ try {
+ return Math.max(Long.parseLong(raw.trim()), 0);
+ } catch (NumberFormatException invalidNumber) {
+ return 0;
+ }
+ }
+
+ private static Integer toInt(long value) {
+ return (int) Math.min(Math.max(value, 0), Integer.MAX_VALUE);
+ }
+
+ /** One parent topic observed across broker masters, with its TTL and
bound consumer groups. */
+ private static final class ParentTopicAccumulator {
+ private final String parentTopic;
+ private String brokerAddr;
+ private final Set<String> groups = new LinkedHashSet<>();
+ private int ttlMinutes = -1;
+
+ private ParentTopicAccumulator(String parentTopic, String brokerAddr) {
+ this.parentTopic = parentTopic;
+ this.brokerAddr = brokerAddr;
+ }
+
+ private void merge(String brokerAddr, Integer ttlMinutes) {
+ this.brokerAddr = brokerAddr;
+ if (ttlMinutes != null && ttlMinutes > this.ttlMinutes) {
+ this.ttlMinutes = ttlMinutes;
+ }
+ }
+ }
+
+ /** The broker master that reported a client's lite subscription, plus
that client's info. */
+ private record LocatedClient(String master, GetLiteClientInfoResponseBody
body) {
+ }
+}
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 e1d6baf8f..f5262dff2 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
@@ -17,66 +17,186 @@
package org.apache.rocketmq.studio.instance.topic;
+import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.model.LiteTopicQuota;
+import org.apache.rocketmq.studio.model.LiteTopicSession;
+import org.apache.rocketmq.studio.model.LiteTopicSummary;
+import org.apache.rocketmq.studio.provider.LiteTopicProvider;
+import org.apache.rocketmq.studio.provider.apache.RocketMQLiteTopicProvider;
+import org.apache.rocketmq.studio.provider.apache.RocketMQProperties;
import org.junit.jupiter.api.Test;
+import java.util.Date;
+import java.util.LinkedHashSet;
+import java.util.List;
+
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;
class LiteTopicServiceTest {
- private final LiteTopicService liteTopicService = new LiteTopicService();
+ /** No NameServer configured: the provider reports the feature as
unavailable. */
+ private final LiteTopicService unavailableService = new LiteTopicService(
+ new RocketMQLiteTopicProvider(new MqAdminExtFactory(), new
RocketMQProperties()));
@Test
- void listLiteTopicsShouldReturnUnsupportedWhenProviderIsUnavailable() {
- assertThatThrownBy(() -> liteTopicService.listLiteTopics("hat", "
DEFAULT "))
+ void listLiteTopicsShouldReportUnsupportedWhenProviderIsUnavailable() {
+ assertThatThrownBy(() -> unavailableService.listLiteTopics("hat", "
DEFAULT "))
.isInstanceOfSatisfying(BusinessException.class, ex -> {
assertThat(ex.getCode()).isEqualTo(501);
- assertThat(ex.getMessage()).isEqualTo("LiteTopic provider
integration is not available");
+
assertThat(ex.getMessage()).isEqualTo(LiteTopicProvider.UNSUPPORTED);
});
}
@Test
- void getQuotaShouldReturnUnsupportedWhenProviderIsUnavailable() {
- assertThatThrownBy(() -> liteTopicService.getQuota("default"))
+ void getQuotaShouldReportUnsupportedWhenProviderIsUnavailable() {
+ assertThatThrownBy(() -> unavailableService.getQuota("default"))
.isInstanceOfSatisfying(BusinessException.class, ex -> {
assertThat(ex.getCode()).isEqualTo(501);
- assertThat(ex.getMessage()).isEqualTo("LiteTopic provider
integration is not available");
+
assertThat(ex.getMessage()).isEqualTo(LiteTopicProvider.UNSUPPORTED);
});
}
@Test
- void getSessionShouldReturnUnsupportedWhenProviderIsUnavailable() {
- assertThatThrownBy(() -> liteTopicService.getSession("sess-001"))
+ void getSessionShouldReportUnsupportedWhenProviderIsUnavailable() {
+ assertThatThrownBy(() ->
unavailableService.getSession("parent~group~client"))
.isInstanceOfSatisfying(BusinessException.class, ex -> {
assertThat(ex.getCode()).isEqualTo(501);
- assertThat(ex.getMessage()).isEqualTo("LiteTopic provider
integration is not available");
+
assertThat(ex.getMessage()).isEqualTo(LiteTopicProvider.UNSUPPORTED);
});
}
@Test
void extendTTLShouldRejectInvalidInput() {
- assertThatThrownBy(() -> liteTopicService.extendTTL("", 1L))
+ assertThatThrownBy(() -> unavailableService.extendTTL("", 1L))
.isInstanceOf(BusinessException.class)
.hasMessage("topicPattern is required")
.satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(400));
- assertThatThrownBy(() ->
liteTopicService.extendTTL("chat/{sessionId}", 0L))
+ assertThatThrownBy(() ->
unavailableService.extendTTL("chat/{sessionId}", 0L))
.isInstanceOf(BusinessException.class)
.hasMessage("newTTL must be positive")
.satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(400));
}
@Test
- void extendTTLShouldReturnUnsupportedWhenProviderIsUnavailable() {
- assertThatThrownBy(() ->
liteTopicService.extendTTL("chat/{sessionId}", 7_200_000L))
+ void extendTTLShouldReportUnsupportedWhenProviderIsUnavailable() {
+ assertThatThrownBy(() ->
unavailableService.extendTTL("chat/{sessionId}", 7_200_000L))
.isInstanceOfSatisfying(BusinessException.class, ex -> {
assertThat(ex.getCode()).isEqualTo(501);
- assertThat(ex.getMessage()).isEqualTo("LiteTopic provider
integration is not available");
+
assertThat(ex.getMessage()).isEqualTo(LiteTopicProvider.UNSUPPORTED);
});
}
+ @Test
+ void getSessionShouldRejectBlankSessionId() {
+ assertThatThrownBy(() -> unavailableService.getSession(" "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("sessionId is required");
+ }
+
@Test
void getCapabilityShouldReportUnsupportedByDefault() {
- assertThat(liteTopicService.getCapability().isSupported()).isFalse();
+ assertThat(unavailableService.getCapability().isSupported()).isFalse();
+ }
+
+ @Test
+ void listLiteTopicsShouldMapProviderModelsOntoViewObjects() {
+ LiteTopicProvider provider = mock(LiteTopicProvider.class);
+ LiteTopicSummary summary = new LiteTopicSummary();
+ summary.setTopicPattern("chat");
+ summary.setNamespace("DEFAULT");
+ summary.setTopicCount(3);
+ summary.setConsumerCount(2);
+ summary.setTotalBacklog(41L);
+ summary.setAverageTTL(1_800_000L);
+ summary.setLastActiveTime(new Date(1_700_000_000_000L));
+ summary.setSessionIds(List.of("chat~g~c1"));
+ when(provider.listLiteTopics("chat",
"DEFAULT")).thenReturn(List.of(summary));
+ when(provider.isSupported()).thenReturn(true);
+
+ LiteTopicService service = new LiteTopicService(provider);
+ List<LiteTopicItemVO> items = service.listLiteTopics("chat",
"DEFAULT");
+
+ assertThat(items).singleElement().satisfies(item -> {
+ assertThat(item.getTopicPattern()).isEqualTo("chat");
+ assertThat(item.getNamespace()).isEqualTo("DEFAULT");
+ assertThat(item.getTopicCount()).isEqualTo(3);
+ assertThat(item.getConsumerCount()).isEqualTo(2);
+ assertThat(item.getTotalBacklog()).isEqualTo(41L);
+ assertThat(item.getAverageTTL()).isEqualTo(1_800_000L);
+ assertThat(item.getLastActiveTime()).isEqualTo(1_700_000_000_000L);
+ assertThat(item.getSessionIds()).containsExactly("chat~g~c1");
+ // lastActiveTime is in the past relative to a 30-minute TTL, so
the session is expired.
+ assertThat(item.getTtlStatus()).isEqualTo("EXPIRED");
+ });
+ assertThat(service.getCapability().isSupported()).isTrue();
+ }
+
+ @Test
+ void getSessionShouldMapProviderModelOntoViewObject() {
+ LiteTopicProvider provider = mock(LiteTopicProvider.class);
+ LiteTopicSession session = new LiteTopicSession();
+ session.setSessionId("chat~g~c1");
+ session.setClientId("c1");
+ session.setClientAddress("10.0.0.9:1234");
+ session.setParentTopic("chat");
+ session.setConsumerGroup("g");
+ session.setTtl(1_800_000L);
+ session.setTtlRemaining(900_000L);
+ session.setStatus("ACTIVE");
+ session.setTotalMessages(120L);
+ session.setConsumedMessages(100L);
+ session.setPendingMessages(20L);
+ session.setLiteTopicCreationCount(2);
+ 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");
+
+ assertThat(vo.getSessionId()).isEqualTo("chat~g~c1");
+ assertThat(vo.getClientAddress()).isEqualTo("10.0.0.9:1234");
+ assertThat(vo.getTtl()).isEqualTo(1_800_000L);
+ assertThat(vo.getTtlRemaining()).isEqualTo(900_000L);
+ assertThat(vo.getTotalMessages()).isEqualTo(120L);
+ assertThat(vo.getConsumedMessages()).isEqualTo(100L);
+ assertThat(vo.getPendingMessages()).isEqualTo(20L);
+ assertThat(vo.getLiteTopicCreationCount()).isEqualTo(2);
+ // Every lite topic inherits the parent topic's TTL policy.
+
assertThat(vo.getLiteTopics()).extracting(LiteTopicSessionVO.SessionLiteTopic::getTopicName)
+ .containsExactly("bob", "alice");
+ assertThat(vo.getLiteTopics()).allSatisfy(row -> {
+ assertThat(row.getStatus()).isEqualTo("ACTIVE");
+ assertThat(row.getTtlRemaining()).isEqualTo(900_000L);
+ });
+ }
+
+ @Test
+ void getQuotaShouldMapProviderModelOntoViewObject() {
+ LiteTopicProvider provider = mock(LiteTopicProvider.class);
+ LiteTopicQuota quota = new LiteTopicQuota();
+ quota.setCurrentTopicCount(10);
+ quota.setMaxTopicCount(40);
+ quota.setCurrentSessionCount(4);
+ quota.setMaxSessionCount(100_000);
+ quota.setDefaultTTL(900_000L);
+ quota.setMaxTTL(2_592_000_000L);
+ quota.setCurrentCreationRate(0.0);
+ quota.setMaxCreationRate(0.0);
+ when(provider.getQuota("DEFAULT")).thenReturn(quota);
+
+ LiteTopicQuotaVO vo = new
LiteTopicService(provider).getQuota("DEFAULT");
+
+ assertThat(vo.getCurrentTopicCount()).isEqualTo(10);
+ assertThat(vo.getMaxTopicCount()).isEqualTo(40);
+ assertThat(vo.getUsageRate()).isEqualTo(0.25);
+ assertThat(vo.getSessionUsageRate()).isCloseTo(0.00004,
org.assertj.core.data.Offset.offset(1e-9));
+ assertThat(vo.getRemainingQuota()).isEqualTo(30);
+ assertThat(vo.getDefaultTTL()).isEqualTo(900_000L);
+ assertThat(vo.getMaxTTL()).isEqualTo(2_592_000_000L);
+ assertThat(vo.getConsumerDensity()).isEqualTo(0.4);
+ assertThat(vo.getCurrentCreationRate()).isZero();
}
}
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
new file mode 100644
index 000000000..58f234e3b
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
@@ -0,0 +1,386 @@
+/*
+ * 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 org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
+import org.apache.rocketmq.common.lite.LiteUtil;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import
org.apache.rocketmq.remoting.protocol.body.GetBrokerLiteInfoResponseBody;
+import
org.apache.rocketmq.remoting.protocol.body.GetLiteClientInfoResponseBody;
+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.common.exception.BusinessException;
+import org.apache.rocketmq.studio.model.LiteTopicQuota;
+import org.apache.rocketmq.studio.model.LiteTopicSession;
+import org.apache.rocketmq.studio.model.LiteTopicSummary;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+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.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class RocketMQLiteTopicProviderTest {
+
+ private static final String NAMESRV = "127.0.0.1:9876";
+ private static final String BROKER_A = "broker-a:10911";
+ private static final String PARENT = "chat";
+ private static final String GROUP = "lite-group";
+
+ private MQAdminExt admin;
+ private RocketMQLiteTopicProvider provider;
+
+ @BeforeEach
+ void setUp() {
+ admin = mock(MQAdminExt.class);
+ MqAdminExtFactory factory = mock(MqAdminExtFactory.class);
+ doAnswer(invocation -> {
+ MqAdminExtFactory.AdminAction<?> action =
invocation.getArgument(2);
+ return action.apply(admin);
+ }).when(factory).execute(anyString(), nullable(RPCHook.class), any());
+ RocketMQProperties properties = new RocketMQProperties();
+ properties.setNamesrvAddr(NAMESRV);
+ provider = new RocketMQLiteTopicProvider(factory, properties);
+ }
+
+ @Test
+ void isSupportedIsFalseWithoutConfiguredNameServer() {
+ assertThat(new
RocketMQLiteTopicProvider(mock(MqAdminExtFactory.class), new
RocketMQProperties())
+ .isSupported()).isFalse();
+ }
+
+ @Test
+ void isSupportedIsFalseWhenBrokerRejectsTheLiteProbe() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.getBrokerLiteInfo(BROKER_A)).thenThrow(new
IllegalStateException("unsupported"));
+
+ assertThat(provider.isSupported()).isFalse();
+ }
+
+ @Test
+ void isSupportedIsTrueWhenBrokerAnswersTheLiteProbe() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.getBrokerLiteInfo(BROKER_A)).thenReturn(brokerLiteInfo(2,
40, 1));
+
+ assertThat(provider.isSupported()).isTrue();
+ }
+
+ @Test
+ void listLiteTopicsAggregatesParentTopicTtlBacklogAndSessions() throws
Exception {
+ long lastAccess = System.currentTimeMillis() - 1_000;
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+
when(admin.getBrokerLiteInfo(BROKER_A)).thenReturn(brokerLiteInfo(PARENT, 30,
2, GROUP));
+ when(admin.getParentTopicInfo(BROKER_A,
PARENT)).thenReturn(parentTopicInfo(PARENT, 30, 2));
+ when(admin.getLiteGroupInfo(BROKER_A, GROUP, null,
1)).thenReturn(lag(7));
+
when(admin.examineConsumerConnectionInfo(GROUP)).thenReturn(consumerConnection("c1",
"10.0.0.9:1234"));
+ when(admin.getLiteClientInfo(BROKER_A, PARENT, GROUP,
"c1")).thenReturn(clientInfo(2, lastAccess));
+
+ List<LiteTopicSummary> summaries = provider.listLiteTopics(null, null);
+
+ assertThat(summaries).singleElement().satisfies(summary -> {
+ assertThat(summary.getTopicPattern()).isEqualTo(PARENT);
+
assertThat(summary.getNamespace()).isEqualTo(RocketMQLiteTopicProvider.DEFAULT_NAMESPACE);
+ assertThat(summary.getTopicCount()).isEqualTo(2);
+ assertThat(summary.getConsumerCount()).isEqualTo(1);
+ assertThat(summary.getTotalBacklog()).isEqualTo(7L);
+
assertThat(summary.getAverageTTL()).isEqualTo(TimeUnit.MINUTES.toMillis(30));
+ assertThat(summary.getSessionIds())
+
.containsExactly(RocketMQLiteTopicProvider.encodeSessionId(PARENT, GROUP,
"c1"));
+ assertThat(summary.getLastActiveTime()).isNotNull();
+ assertThat(summary.getTTLStatus()).isEqualTo("ACTIVE");
+ });
+ }
+
+ @Test
+ void listLiteTopicsFiltersByPatternCaseInsensitively() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+
when(admin.getBrokerLiteInfo(BROKER_A)).thenReturn(brokerLiteInfo(PARENT, 30,
1, GROUP));
+ when(admin.getParentTopicInfo(BROKER_A,
PARENT)).thenReturn(parentTopicInfo(PARENT, 30, 1));
+ when(admin.getLiteGroupInfo(BROKER_A, GROUP, null,
1)).thenReturn(lag(0));
+ when(admin.examineConsumerConnectionInfo(GROUP)).thenReturn(new
ConsumerConnection());
+ when(admin.getLiteClientInfo(anyString(), anyString(), anyString(),
anyString()))
+ .thenReturn(clientInfo(1, System.currentTimeMillis()));
+
+ assertThat(provider.listLiteTopics("CHAT", null)).hasSize(1);
+ assertThat(provider.listLiteTopics("orders", null)).isEmpty();
+ assertThat(provider.listLiteTopics(null, "default")).hasSize(1);
+ assertThat(provider.listLiteTopics(null, "other")).isEmpty();
+ }
+
+ @Test
+ void getSessionResolvesTheBrokerOwningTheClientAndComputesProgress()
throws Exception {
+ long lastAccess = System.currentTimeMillis() - 1_000;
+ Map<String, BrokerData> brokers = new LinkedHashMap<>();
+ brokers.put("broker-a", broker(BROKER_A));
+ brokers.put("broker-b", broker("broker-b:10911"));
+ ClusterInfo clusterInfo = new ClusterInfo();
+ clusterInfo.setBrokerAddrTable(brokers);
+ when(admin.examineBrokerClusterInfo()).thenReturn(clusterInfo);
+ // The first master does not hold this client's lite subscription; the
second one does.
+ when(admin.getLiteClientInfo(BROKER_A, PARENT, GROUP,
"c1")).thenReturn(clientInfo(-1, 0));
+ when(admin.getLiteClientInfo("broker-b:10911", PARENT, GROUP, "c1"))
+ .thenReturn(clientInfo(1, lastAccess,
LiteUtil.toLmqName(PARENT, "bob")));
+ when(admin.getParentTopicInfo("broker-b:10911",
PARENT)).thenReturn(parentTopicInfo(PARENT, 30, 1));
+ when(admin.getLiteGroupInfo("broker-b:10911", GROUP, null,
1)).thenReturn(lag(5));
+ when(admin.getLiteGroupInfo("broker-b:10911", GROUP, "bob",
1)).thenReturn(consumed(10, 10));
+
when(admin.examineConsumerConnectionInfo(GROUP)).thenReturn(consumerConnection("c1",
"10.0.0.9:1234"));
+
+ LiteTopicSession session = provider.getSession(
+ RocketMQLiteTopicProvider.encodeSessionId(PARENT, GROUP,
"c1"));
+
+ assertThat(session.getSessionId()).isEqualTo("chat~lite-group~c1");
+ assertThat(session.getClientId()).isEqualTo("c1");
+ assertThat(session.getClientAddress()).isEqualTo("10.0.0.9:1234");
+ assertThat(session.getParentTopic()).isEqualTo(PARENT);
+ assertThat(session.getConsumerGroup()).isEqualTo(GROUP);
+ assertThat(session.getTtl()).isEqualTo(TimeUnit.MINUTES.toMillis(30));
+ assertThat(session.getTtlRemaining()).isGreaterThan(0);
+ assertThat(session.getStatus()).isEqualTo("ACTIVE");
+ assertThat(session.getLiteTopicCreationCount()).isEqualTo(1);
+ assertThat(session.getLiteTopics()).containsExactly("bob");
+ // pending lag (5) plus the committed offset read back from the lite
topic (10).
+ assertThat(session.getPendingMessages()).isEqualTo(5L);
+ assertThat(session.getConsumedMessages()).isEqualTo(10L);
+ assertThat(session.getTotalMessages()).isEqualTo(15L);
+ }
+
+ @Test
+ void getSessionFailsWhenNoBrokerReportsTheClient() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.getLiteClientInfo(anyString(), anyString(), anyString(),
anyString()))
+ .thenReturn(clientInfo(-1, 0));
+
+ assertThatThrownBy(() -> provider.getSession("chat~lite-group~c1"))
+ .isInstanceOfSatisfying(BusinessException.class,
+ ex -> assertThat(ex.getCode()).isEqualTo(404));
+ }
+
+ @Test
+ void getSessionRejectsMalformedSessionId() {
+ assertThatThrownBy(() -> provider.getSession("not-a-session"))
+ .isInstanceOfSatisfying(BusinessException.class,
+ ex -> assertThat(ex.getCode()).isEqualTo(400));
+ }
+
+ @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));
+
+ ArgumentCaptor<TopicConfig> captor =
ArgumentCaptor.forClass(TopicConfig.class);
+ verify(admin).createAndUpdateTopicConfig(eq(BROKER_A),
captor.capture());
+ // The update must be expressed in the broker's "+key=value" change
protocol and must not
+ // re-send the immutable message.type attribute.
+ Map<String, String> change = captor.getValue().getAttributes();
+ assertThat(change).containsEntry("+lite.topic.expiration", "120");
+ assertThat(change).hasSize(1);
+ }
+
+ @Test
+ void extendTtlClampsToTheProtocolMaximum() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.examineTopicConfig(BROKER_A,
PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
+
+ provider.extendTTL(PARENT, TimeUnit.DAYS.toMillis(90));
+
+ ArgumentCaptor<TopicConfig> captor =
ArgumentCaptor.forClass(TopicConfig.class);
+ verify(admin).createAndUpdateTopicConfig(eq(BROKER_A),
captor.capture());
+ assertThat(captor.getValue().getAttributes())
+ .containsEntry("+lite.topic.expiration",
+
String.valueOf(RocketMQLiteTopicProvider.MAX_LITE_TTL_MINUTES));
+ }
+
+ @Test
+ void extendTtlFailsForANonLiteTopic() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.examineTopicConfig(BROKER_A, PARENT)).thenReturn(new
TopicConfig(PARENT));
+
+ assertThatThrownBy(() -> provider.extendTTL(PARENT, 60_000L))
+ .isInstanceOfSatisfying(BusinessException.class,
+ ex -> assertThat(ex.getCode()).isEqualTo(404));
+ verify(admin, never()).createAndUpdateTopicConfig(anyString(), any());
+ }
+
+ @Test
+ void getQuotaAggregatesBrokerLiteCapacityAndConfigLimits() throws
Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.getBrokerLiteInfo(BROKER_A)).thenReturn(brokerLiteInfo(3,
40, 3));
+ Properties brokerConfig = new Properties();
+ brokerConfig.setProperty("maxLiteSubscriptionCount", "100000");
+ brokerConfig.setProperty("minLiteTTl", "900000");
+ when(admin.getBrokerConfig(BROKER_A)).thenReturn(brokerConfig);
+
+ LiteTopicQuota quota = provider.getQuota(null);
+
+ assertThat(quota.getCurrentTopicCount()).isEqualTo(3);
+ assertThat(quota.getMaxTopicCount()).isEqualTo(40);
+ assertThat(quota.getCurrentSessionCount()).isEqualTo(3);
+ assertThat(quota.getMaxSessionCount()).isEqualTo(100_000);
+ // minLiteTTl is the broker's floor, not a namespace default: it must
not be surfaced
+ // as defaultTTL, which the console would render as the TTL applied to
new lite topics.
+ assertThat(quota.getDefaultTTL()).isNull();
+ assertThat(quota.getMaxTTL()).isEqualTo(TimeUnit.MINUTES.toMillis(
+ RocketMQLiteTopicProvider.MAX_LITE_TTL_MINUTES));
+ assertThat(quota.getRemainingQuota()).isEqualTo(37);
+ }
+
+ @Test
+ void quotaSkipsMastersWhoseLiteInfoIsUnavailable() throws Exception {
+ String silentMaster = "127.0.0.1:10912";
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A,
silentMaster));
+ when(admin.getBrokerLiteInfo(BROKER_A)).thenReturn(brokerLiteInfo(3,
40, 3));
+ when(admin.getBrokerLiteInfo(silentMaster)).thenReturn(null);
+ Properties reachableConfig = new Properties();
+ reachableConfig.setProperty("maxLiteSubscriptionCount", "100000");
+ when(admin.getBrokerConfig(BROKER_A)).thenReturn(reachableConfig);
+ Properties silentConfig = new Properties();
+ silentConfig.setProperty("maxLiteSubscriptionCount", "100000");
+ when(admin.getBrokerConfig(silentMaster)).thenReturn(silentConfig);
+
+ LiteTopicQuota quota = provider.getQuota(null);
+
+ // Both masters advertise the same cap but only one contributed
current counts, so the
+ // ratio has to be built from that master alone rather than mixing the
two master sets.
+ assertThat(quota.getMaxSessionCount()).isEqualTo(100_000);
+ assertThat(quota.getCurrentSessionCount()).isEqualTo(3);
+ assertThat(quota.getMaxTopicCount()).isEqualTo(40);
+ }
+
+ @Test
+ void quotaFailsWhenNoBrokerMasterIsReachable() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(new ClusterInfo());
+
+ assertThatThrownBy(() -> provider.getQuota(null))
+ .isInstanceOfSatisfying(BusinessException.class,
+ ex -> assertThat(ex.getCode()).isEqualTo(503));
+ }
+
+ // ─── Fixtures ─────────────────────────────────────────────────────
+
+ private static ClusterInfo cluster(String... masterAddresses) {
+ Map<String, BrokerData> table = new LinkedHashMap<>();
+ for (String address : masterAddresses) {
+ table.put(address, broker(address));
+ }
+ ClusterInfo clusterInfo = new ClusterInfo();
+ clusterInfo.setBrokerAddrTable(table);
+ return clusterInfo;
+ }
+
+ private static BrokerData broker(String masterAddress) {
+ BrokerData data = new BrokerData();
+ data.setBrokerAddrs(new HashMap<>(Map.of(0L, masterAddress)));
+ return data;
+ }
+
+ private static GetBrokerLiteInfoResponseBody brokerLiteInfo(int
currentLmq, int maxLmq,
+ int
liteSubscriptions) {
+ GetBrokerLiteInfoResponseBody body = new
GetBrokerLiteInfoResponseBody();
+ body.setCurrentLmqNum(currentLmq);
+ body.setMaxLmqNum(maxLmq);
+ body.setLiteSubscriptionCount(liteSubscriptions);
+ return body;
+ }
+
+ private static GetBrokerLiteInfoResponseBody brokerLiteInfo(String
parentTopic, int ttlMinutes,
+ int
currentLmq, String group) {
+ GetBrokerLiteInfoResponseBody body = brokerLiteInfo(currentLmq, 40, 0);
+ Map<String, Integer> topicMeta = new LinkedHashMap<>();
+ topicMeta.put(parentTopic, ttlMinutes);
+ body.setTopicMeta(topicMeta);
+ Map<String, Set<String>> groupMeta = new LinkedHashMap<>();
+ groupMeta.put(parentTopic, new HashSet<>(Set.of(group)));
+ body.setGroupMeta(groupMeta);
+ return body;
+ }
+
+ private static GetParentTopicInfoResponseBody parentTopicInfo(String
topic, int ttl, int liteTopicCount) {
+ GetParentTopicInfoResponseBody body = new
GetParentTopicInfoResponseBody();
+ body.setTopic(topic);
+ body.setTtl(ttl);
+ body.setLiteTopicCount(liteTopicCount);
+ return body;
+ }
+
+ private static GetLiteGroupInfoResponseBody lag(long totalLag) {
+ GetLiteGroupInfoResponseBody body = new GetLiteGroupInfoResponseBody();
+ body.setTotalLagCount(totalLag);
+ return body;
+ }
+
+ private static GetLiteGroupInfoResponseBody consumed(long brokerOffset,
long consumerOffset) {
+ OffsetWrapper wrapper = new OffsetWrapper();
+ wrapper.setBrokerOffset(brokerOffset);
+ wrapper.setConsumerOffset(consumerOffset);
+ GetLiteGroupInfoResponseBody body = new GetLiteGroupInfoResponseBody();
+ body.setLiteTopicOffsetWrapper(wrapper);
+ return body;
+ }
+
+ private static GetLiteClientInfoResponseBody clientInfo(int
liteTopicCount, long lastAccessTime,
+ String...
lmqNames) {
+ GetLiteClientInfoResponseBody body = new
GetLiteClientInfoResponseBody();
+ body.setLiteTopicCount(liteTopicCount);
+ body.setLastAccessTime(lastAccessTime);
+ body.setLiteTopicSet(new HashSet<>(Set.of(lmqNames)));
+ return body;
+ }
+
+ private static ConsumerConnection consumerConnection(String clientId,
String clientAddr) {
+ Connection connection = new Connection();
+ connection.setClientId(clientId);
+ connection.setClientAddr(clientAddr);
+ ConsumerConnection consumerConnection = new ConsumerConnection();
+ consumerConnection.setConnectionSet(new HashSet<>(Set.of(connection)));
+ return consumerConnection;
+ }
+
+ private static TopicConfig liteTopicConfig(String topic, int ttlMinutes) {
+ TopicConfig config = new TopicConfig(topic);
+ config.setTopicMessageType(TopicMessageType.LITE);
+ config.setLiteTopicExpiration(ttlMinutes);
+ return config;
+ }
+}
