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 bd12f98fe fix(litetopic): report partial multi-master TTL updates
instead of a bare write error (#7137)
bd12f98fe is described below
commit bd12f98fe801ee7eb9511ef527f03f377aa362e5
Author: btlqql <[email protected]>
AuthorDate: Mon Oct 26 20:35:00 2026 +0800
fix(litetopic): report partial multi-master TTL updates instead of a bare
write error (#7137)
* fix(litetopic): report partial multi-master TTL updates instead of a bare
write error (#7085)
extendTTL writes lite.topic.expiration to every broker master that carries
the Lite parent
topic. When a later write failed, the earlier masters kept the new
expiration but the caller
only received the raw write error, so a cluster could stay split between
the old and the new
TTL with nothing in the response saying which masters had been changed.
The write phase now tracks the masters it already updated and, on a failure
after the first
one, throws a 502 BusinessException that names the updated masters, the
failed master with its
error and the masters that were not attempted, plus the recovery path: the
extension is
idempotent, so re-running it converges every master. A failure on the first
write still
propagates unchanged and the all-master success path is untouched.
---
.../provider/apache/RocketMQLiteTopicProvider.java | 46 ++++++++++++++++++++--
.../apache/RocketMQLiteTopicProviderTest.java | 43 ++++++++++++++++++++
2 files changed, 86 insertions(+), 3 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 06ccfb960..478988355 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
@@ -396,7 +396,11 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
if (pending.isEmpty()) {
throw new BusinessException(404, "Lite parent topic not found:
" + topicPattern);
}
- for (Map.Entry<String, TopicConfig> entry : pending.entrySet()) {
+ List<String> masters = new ArrayList<>(pending.keySet());
+ List<String> updatedMasters = new ArrayList<>();
+ for (int index = 0; index < masters.size(); index++) {
+ String master = masters.get(index);
+ TopicConfig config = pending.get(master);
// 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
@@ -405,8 +409,23 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
// Only the TTL is altered.
Map<String, String> change = new HashMap<>();
change.put("+lite.topic.expiration", String.valueOf(minutes));
- entry.getValue().setAttributes(change);
- admin.createAndUpdateTopicConfig(entry.getKey(),
entry.getValue());
+ config.setAttributes(change);
+ try {
+ admin.createAndUpdateTopicConfig(master, config);
+ } catch (Exception failure) {
+ restoreInterrupt(failure);
+ if (updatedMasters.isEmpty()) {
+ // No master was touched yet, so the cluster still
holds the previous
+ // expiration and the caller keeps the plain write
error.
+ throw failure;
+ }
+ // The masters listed above already store the new
expiration while this one
+ // does not: report that split - and the way out of it -
instead of a bare
+ // write error that hides which masters the request
already changed.
+ throw partialTtlUpdate(topicPattern, minutes,
updatedMasters, master,
+ masters.subList(index + 1, masters.size()),
failure);
+ }
+ updatedMasters.add(master);
}
log.info("Extended LiteTopic TTL to {}ms ({} min) for parent topic
{} on {} broker(s)",
ttlMillis, minutes, topicPattern, pending.size());
@@ -414,6 +433,27 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
}));
}
+ /**
+ * Reports a TTL write phase that stopped after at least one master was
rewritten: the caller
+ * needs to know which masters already store the new expiration, which one
failed and which ones
+ * were not attempted. The extension is idempotent, so re-running it is
the recovery path the
+ * message names.
+ */
+ private static BusinessException partialTtlUpdate(String topicPattern,
long minutes,
+ List<String>
updatedMasters, String failedMaster,
+ List<String>
notUpdatedMasters,
+ Exception failure) {
+ String untouched = notUpdatedMasters.isEmpty()
+ ? "every other master was already updated"
+ : "not updated yet: " + String.join(", ", notUpdatedMasters);
+ return new BusinessException(502, "LiteTopic TTL for " + topicPattern
+ + " is only partially applied: " + minutes + " minute(s)
reached "
+ + String.join(", ", updatedMasters) + " before " +
failedMaster + " failed ("
+ + failure.getMessage() + "); " + untouched
+ + ". Re-run the same TTL extension for " + topicPattern
+ + " to apply it to every master; the write is idempotent.");
+ }
+
/**
* 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)
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 b3b644ef2..f0b9bcf56 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
@@ -61,6 +61,7 @@ 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.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -464,6 +465,48 @@ class RocketMQLiteTopicProviderTest {
verify(admin, never()).createAndUpdateTopicConfig(eq(peerMaster),
any());
}
+ @Test
+ void
extendTtlReportsAPartialWriteInsteadOfABareFailureWhenALaterMasterFails()
throws Exception {
+ String peerMaster = "broker-b:10911";
+ String untouchedMaster = "broker-c:10911";
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A,
peerMaster, untouchedMaster));
+ when(admin.examineTopicConfig(BROKER_A,
PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
+ when(admin.examineTopicConfig(peerMaster,
PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
+ when(admin.examineTopicConfig(untouchedMaster,
PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
+ doThrow(new RemotingTimeoutException("broker-b write timed out"))
+ .when(admin).createAndUpdateTopicConfig(eq(peerMaster), any());
+
+ // broker-a already stores the new expiration while broker-b keeps the
old one, so the
+ // caller has to see that split and the way out of it, not just the
raw write error.
+ assertThatThrownBy(() -> provider.extendTTL("cluster-a", PARENT,
TimeUnit.MINUTES.toMillis(120)))
+ .isInstanceOfSatisfying(BusinessException.class, ex -> {
+ assertThat(ex.getCode()).isEqualTo(502);
+ assertThat(ex.getMessage())
+ .contains("120 minute(s) reached " + BROKER_A)
+ .contains("before " + peerMaster + " failed
(broker-b write timed out)")
+ .contains("not updated yet: " + untouchedMaster)
+ .contains("Re-run the same TTL extension for " +
PARENT);
+ });
+ verify(admin).createAndUpdateTopicConfig(eq(BROKER_A),
any(TopicConfig.class));
+ verify(admin, never()).createAndUpdateTopicConfig(eq(untouchedMaster),
any());
+ }
+
+ @Test
+ void extendTtlKeepsTheBareWriteFailureWhenNoMasterWasUpdatedYet() throws
Exception {
+ String peerMaster = "broker-b:10911";
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A,
peerMaster));
+ when(admin.examineTopicConfig(BROKER_A,
PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
+ when(admin.examineTopicConfig(peerMaster,
PARENT)).thenReturn(liteTopicConfig(PARENT, 30));
+ doThrow(new RemotingTimeoutException("broker-a write timed out"))
+ .when(admin).createAndUpdateTopicConfig(eq(BROKER_A), any());
+
+ // The first write failed, so the cluster is still on the previous
expiration: the caller
+ // keeps the plain write error and the remaining masters are not
attempted.
+ assertThatThrownBy(() -> provider.extendTTL("cluster-a", PARENT,
TimeUnit.MINUTES.toMillis(120)))
+ .isInstanceOf(RemotingTimeoutException.class);
+ verify(admin, never()).createAndUpdateTopicConfig(eq(peerMaster),
any());
+ }
+
@Test
void getQuotaAggregatesBrokerLiteCapacityAndConfigLimits() throws
Exception {
when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
