This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 4f0cd94d98 [ISSUE #11486] Prevent evicted LiteTopic clients from
rejoining dispatch on full sync (#11487)
4f0cd94d98 is described below
commit 4f0cd94d981b0f501c317aa215c76072c1ab6180
Author: Xiao Yang <[email protected]>
AuthorDate: Tue Oct 27 10:35:56 2026 +0800
[ISSUE #11486] Prevent evicted LiteTopic clients from rejoining dispatch on
full sync (#11487)
---
.../broker/lite/LiteSubscriptionRegistryImpl.java | 7 +++++--
.../broker/lite/LiteSubscriptionRegistryImplTest.java | 15 ++++++++++++++-
2 files changed, 19 insertions(+), 3 deletions(-)
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java
b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java
index b818ff29cb..115add8739 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java
@@ -144,15 +144,18 @@ public class LiteSubscriptionRegistryImpl extends
ServiceThread implements LiteS
Set<String> lmqsToRemove =
LiteSubscription.removals(thisSub.getLmqSet(), targetLmqs);
// 3) Apply removals, then idempotent adds over the full target set
+ boolean isExclusiveGroup = LiteMetadataUtil.isSubLiteExclusive(group,
brokerController);
ClientGroup clientGroup = new ClientGroup(clientId,
thisSub.getGroup());
lmqsToRemove.forEach(lmqName -> {
thisSub.removeLmq(lmqName);
removeClientGroup(clientGroup, lmqName, false);
});
- targetLmqs.forEach(lmqName -> addClientGroup(thisSub, clientGroup,
lmqName));
+ targetLmqs.stream()
+ .filter(lmqName -> !isExclusiveGroup ||
!exclusiveEvictionTombstones.contains(clientId, lmqName))
+ .forEach(lmqName -> addClientGroup(thisSub, clientGroup, lmqName));
// Tombstone operations only apply to exclusive groups.
- if (LiteMetadataUtil.isSubLiteExclusive(group, brokerController)) {
+ if (isExclusiveGroup) {
// If any lmqName in the client's full subscription still has a
tombstone,
// the previous notifyUnsubscribeLite was likely lost. Re-send to
drive convergence.
targetLmqs.stream()
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
index 4fc94e150f..212194ec57 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
@@ -808,6 +808,10 @@ public class LiteSubscriptionRegistryImplTest {
assertFalse(registry.hasExclusiveEvictionTombstone(clientA, "lmq1"));
// cleared (not in newActiveSet)
assertTrue(registry.hasExclusiveEvictionTombstone(clientA, "lmq2"));
// retained (in newActiveSet)
+ assertEquals(Collections.singleton("lmq3"),
registry.getLiteSubscription(clientA).getLmqSet());
+ assertEquals(Collections.singletonList(new ClientGroup(clientB,
group)),
+ registry.getAllSubscribers(group, "lmq2").get(group));
+ assertEquals(3, registry.getActiveSubscriptionNum());
}
/**
@@ -913,10 +917,14 @@ public class LiteSubscriptionRegistryImplTest {
assertTrue(registry.hasExclusiveEvictionTombstone(clientA, lmqName));
- // clientA re-claims the lmqName → tombstone should be cleared
+ registry.addCompleteSubscription(clientA, group, topic, lmqNameSet,
1L);
+ // clientA explicitly re-claims the lmqName → tombstone should be
cleared
registry.addPartialSubscription(clientA, group, topic, lmqNameSet,
null);
assertFalse(registry.hasExclusiveEvictionTombstone(clientA, lmqName));
+ assertEquals(Collections.singletonList(new ClientGroup(clientA,
group)),
+ registry.getAllSubscribers(group, lmqName).get(group));
+ assertEquals(1, registry.getActiveSubscriptionNum());
}
/**
@@ -951,6 +959,11 @@ public class LiteSubscriptionRegistryImplTest {
fullSet.add(lmqName);
registry.addCompleteSubscription(clientA, group, topic, fullSet, 2L);
+
assertTrue(registry.getLiteSubscription(clientA).getLmqSet().isEmpty());
+ assertEquals(Collections.singletonList(new ClientGroup(clientB,
group)),
+ registry.getAllSubscribers(group, lmqName).get(group));
+ assertEquals(1, registry.getActiveSubscriptionNum());
+
// Verify notifyUnsubscribeLite was called for the tombstoned lmqName
during full sync
// The first call is during eviction, the second is during re-notify
ArgumentCaptor<NotifyUnsubscribeLiteRequestHeader> captor =
