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 5479af6f4 fix(lite): fail the TTL extension instead of partially
updating the cluster (#4578)
5479af6f4 is described below
commit 5479af6f4d6203361c78e0b37581d169ad8d6bf8
Author: Zhao Jianing <[email protected]>
AuthorDate: Mon Sep 21 21:02:57 2026 +0800
fix(lite): fail the TTL extension instead of partially updating the cluster
(#4578)
`RocketMQLiteTopicProvider.extendTTL` read and wrote each master inside one
loop, and its private `liteTopicConfig` helper caught every exception and
returned null, which the caller treated as "nothing to extend here" and
skipped. A master that timed out or was mid-restart was left out of the write
while `updated` still counted the others, so `POST /api/lite-topic/extendTTL`
returned 200 and logged the broker count while `lite.topic.expiration` diverged
across the cluster — and becaus [...]
The loop is now two phases: every master's `TopicConfig` is read into a
`LinkedHashMap` first, and only then is the new expiration written back. The
helper was renamed `liteParentTopicConfig`, declares `throws Exception`, and
treats only `MQBrokerException` with `TOPIC_NOT_EXIST`, a null config or a
non-LITE topic as a legitimate skip; the "nothing to update" 404 moved from
`updated == 0` to `pending.isEmpty()`, so it fires before any write.
A failure during the write phase still leaves earlier masters updated —
inherent to a non-transactional multi-broker write — but it now surfaces as an
error rather than a success, and the update is idempotent, so a retry converges.
---
.../provider/apache/RocketMQLiteTopicProvider.java | 52 +++++++++++++++-------
.../apache/RocketMQLiteTopicProviderTest.java | 32 +++++++++++++
2 files changed, 67 insertions(+), 17 deletions(-)
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 f7b160744..16fdd5ceb 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
@@ -16,10 +16,12 @@
*/
package org.apache.rocketmq.studio.provider.apache;
+import org.apache.rocketmq.client.exception.MQBrokerException;
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.ResponseCode;
import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.Connection;
@@ -339,12 +341,21 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
}
long minutes = Math.min(Math.max(Math.round(ttlMillis / 60000.0), 1),
MAX_LITE_TTL_MINUTES);
execute(admin -> {
- int updated = 0;
+ // Read every master's config before writing any of them: a master
that cannot be
+ // examined must fail the request up front instead of being
silently skipped, which
+ // would leave the cluster with mixed lite.topic.expiration
attributes while the
+ // console reports a fully successful extension.
+ Map<String, TopicConfig> pending = new LinkedHashMap<>();
for (String master : masterAddresses(admin)) {
- TopicConfig config = liteTopicConfig(admin, master,
topicPattern);
- if (config == null) {
- continue;
+ TopicConfig config = liteParentTopicConfig(admin, master,
topicPattern);
+ if (config != null) {
+ pending.put(master, config);
}
+ }
+ if (pending.isEmpty()) {
+ throw new BusinessException(404, "Lite parent topic not found:
" + topicPattern);
+ }
+ for (Map.Entry<String, TopicConfig> entry : pending.entrySet()) {
// 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
@@ -353,30 +364,37 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
// 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);
+ entry.getValue().setAttributes(change);
+ admin.createAndUpdateTopicConfig(entry.getKey(),
entry.getValue());
}
log.info("Extended LiteTopic TTL to {}ms ({} min) for parent topic
{} on {} broker(s)",
- ttlMillis, minutes, topicPattern, updated);
+ ttlMillis, minutes, topicPattern, pending.size());
return null;
});
}
- private TopicConfig liteTopicConfig(MQAdminExt admin, String brokerAddr,
String topic) {
+ /**
+ * Reads the parent topic's config for the TTL update loop, distinguishing
the outcomes the
+ * loop must treat differently: a topic that is absent on this master (or
not a LITE topic)
+ * is a legitimate skip, while any other read failure propagates so the
update cannot be
+ * applied to only part of the cluster and still report success.
+ */
+ private TopicConfig liteParentTopicConfig(MQAdminExt admin, String
brokerAddr, String topic)
+ throws Exception {
+ TopicConfig config;
try {
- TopicConfig config = admin.examineTopicConfig(brokerAddr, topic);
- if (config == null ||
!TopicMessageType.LITE.equals(config.getTopicMessageType())) {
+ config = admin.examineTopicConfig(brokerAddr, topic);
+ } catch (MQBrokerException failure) {
+ if (failure.getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
+ log.debug("Parent topic {} is not configured on {}", topic,
brokerAddr);
return null;
}
- return config;
- } catch (Exception failure) {
- log.debug("Parent topic {} is not configured on {}: {}", topic,
brokerAddr, failure.getMessage());
+ throw failure;
+ }
+ if (config == null ||
!TopicMessageType.LITE.equals(config.getTopicMessageType())) {
return null;
}
+ return config;
}
// ─── Quota ────────────────────────────────────────────────────────
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
index 0fbf4c004..e8b213d35 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
@@ -16,10 +16,13 @@
*/
package org.apache.rocketmq.studio.provider.apache;
+import org.apache.rocketmq.client.exception.MQBrokerException;
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.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.Connection;
@@ -301,6 +304,35 @@ class RocketMQLiteTopicProviderTest {
verify(admin, never()).createAndUpdateTopicConfig(anyString(), any());
}
+ @Test
+ void extendTtlFailsInsteadOfPartiallyUpdatingWhenAMasterCannotBeRead()
throws Exception {
+ String unreachableMaster = "127.0.0.1:10912";
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A,
unreachableMaster));
+ when(admin.examineTopicConfig(BROKER_A,
PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
+ when(admin.examineTopicConfig(unreachableMaster, PARENT))
+ .thenThrow(new RemotingTimeoutException("broker restarting"));
+
+ // The extension must fail before any master is written: the
alternative is a cluster
+ // with mixed lite.topic.expiration attributes and a console that
reports success.
+ assertThatThrownBy(() -> provider.extendTTL(PARENT,
TimeUnit.MINUTES.toMillis(120)))
+ .isInstanceOf(RemotingTimeoutException.class);
+ verify(admin, never()).createAndUpdateTopicConfig(anyString(), any());
+ }
+
+ @Test
+ void extendTtlStillUpdatesReachableMastersWhenTheTopicIsAbsentOnAPeer()
throws Exception {
+ String peerMaster = "127.0.0.1:10912";
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A,
peerMaster));
+ when(admin.examineTopicConfig(BROKER_A,
PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
+ when(admin.examineTopicConfig(peerMaster, PARENT))
+ .thenThrow(new MQBrokerException(ResponseCode.TOPIC_NOT_EXIST,
"topic not exist"));
+
+ provider.extendTTL(PARENT, TimeUnit.MINUTES.toMillis(120));
+
+ verify(admin).createAndUpdateTopicConfig(eq(BROKER_A),
any(TopicConfig.class));
+ verify(admin, never()).createAndUpdateTopicConfig(eq(peerMaster),
any());
+ }
+
@Test
void getQuotaAggregatesBrokerLiteCapacityAndConfigLimits() throws
Exception {
when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
