This is an automated email from the ASF dual-hosted git repository.
lizhimins 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 8227ecd1ac [ISSUE #11170] Preserve transaction subscriptions before
first heartbeat (#11171)
8227ecd1ac is described below
commit 8227ecd1acab927d3400b115644a60f489ff6169
Author: qianye <[email protected]>
AuthorDate: Mon Sep 21 16:01:43 2026 +0800
[ISSUE #11170] Preserve transaction subscriptions before first heartbeat
(#11171)
---
.../transaction/ClusterTransactionService.java | 13 ++-
.../transaction/ClusterTransactionServiceTest.java | 123 +++++++++++++++++++++
2 files changed, 135 insertions(+), 1 deletion(-)
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java
index 1ec4286463..dd1ebc526b 100644
---
a/proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java
+++
b/proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java
@@ -82,6 +82,10 @@ public class ClusterTransactionService extends
AbstractTransactionService {
clusterDataSet = Sets.newHashSet();
}
clusterDataSet.addAll(getClusterDataFromTopic(ctx, topic));
+ long now = System.currentTimeMillis();
+ for (ClusterData clusterData : clusterDataSet) {
+ clusterData.lastActiveTimestamp = now;
+ }
return clusterDataSet;
});
} catch (Exception e) {
@@ -131,14 +135,19 @@ public class ClusterTransactionService extends
AbstractTransactionService {
if (clusterDataSet.isEmpty()) {
return null;
}
+ long now = System.currentTimeMillis();
if (!this.producerManager.groupOnline(groupName)) {
- return null;
+ // A transaction send may precede the producer's first
heartbeat.
+ long timeoutMillis =
ConfigurationManager.getProxyConfig().getChannelExpiredTimeout();
+ clusterDataSet.removeIf(clusterData -> now -
clusterData.lastActiveTimestamp >= timeoutMillis);
+ return clusterDataSet.isEmpty() ? null : clusterDataSet;
}
ProducerData producerData = new ProducerData();
producerData.setGroupName(groupName);
for (ClusterData clusterData : clusterDataSet) {
+ clusterData.lastActiveTimestamp = now;
List<HeartbeatData> heartbeatDataList =
clusterHeartbeatData.get(clusterData.cluster);
if (heartbeatDataList == null) {
heartbeatDataList = new ArrayList<>();
@@ -228,6 +237,8 @@ public class ClusterTransactionService extends
AbstractTransactionService {
static class ClusterData {
private final String cluster;
+ // Epoch milliseconds, refreshed on subscription updates or when the
owning group is observed online.
+ private long lastActiveTimestamp = System.currentTimeMillis();
public ClusterData(String cluster) {
this.cluster = cluster;
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java
index 91af74cbef..fdbe3f3e92 100644
---
a/proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java
+++
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java
@@ -17,12 +17,16 @@
package org.apache.rocketmq.proxy.service.transaction;
+import java.lang.reflect.Field;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.rocketmq.broker.client.ProducerManager;
import org.apache.rocketmq.common.MixAll;
@@ -44,6 +48,7 @@ import org.mockito.Mockito;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -108,6 +113,124 @@ public class ClusterTransactionServiceTest extends
BaseServiceTest {
assertEquals(0,
this.clusterTransactionService.getGroupClusterData().size());
}
+ @Test
+ public void testTransactionSubscriptionSurvivesFirstHeartbeatDelay() {
+ List<HeartbeatData> heartbeats = recordHeartbeats();
+ this.clusterTransactionService.addTransactionSubscription(ctx, GROUP,
TOPIC);
+
+ this.clusterTransactionService.scanProducerHeartBeat();
+
assertTrue(this.clusterTransactionService.getGroupClusterData().containsKey(GROUP));
+ assertTrue(heartbeats.isEmpty());
+
+ when(this.producerManager.groupOnline(GROUP)).thenReturn(true);
+ this.clusterTransactionService.scanProducerHeartBeat();
+
+ assertEquals(1, heartbeats.size());
+ assertEquals(GROUP,
heartbeats.get(0).getProducerDataSet().iterator().next().getGroupName());
+ }
+
+ @Test
+ public void testNeverRegisteredSubscriptionExpiresWithoutHeartbeats()
throws Exception {
+ List<HeartbeatData> heartbeats = recordHeartbeats();
+ this.clusterTransactionService.addTransactionSubscription(ctx, GROUP,
TOPIC);
+
+ this.clusterTransactionService.scanProducerHeartBeat();
+ this.clusterTransactionService.scanProducerHeartBeat();
+
assertTrue(this.clusterTransactionService.getGroupClusterData().containsKey(GROUP));
+ assertTrue(heartbeats.isEmpty());
+
+ expireSubscriptions();
+ this.clusterTransactionService.scanProducerHeartBeat();
+
+
assertFalse(this.clusterTransactionService.getGroupClusterData().containsKey(GROUP));
+ assertTrue(heartbeats.isEmpty());
+ }
+
+ @Test
+ public void testAddingExistingSubscriptionRefreshesGracePeriod() throws
Exception {
+ List<HeartbeatData> heartbeats = recordHeartbeats();
+ this.clusterTransactionService.addTransactionSubscription(ctx, GROUP,
TOPIC);
+ expireSubscriptions();
+
+ // Set.addAll keeps the existing ClusterData when another topic uses
the same cluster.
+ this.clusterTransactionService.addTransactionSubscription(ctx, GROUP,
TOPIC + "-another");
+ this.clusterTransactionService.scanProducerHeartBeat();
+
+ assertEquals(1,
this.clusterTransactionService.getGroupClusterData().get(GROUP).size());
+ assertTrue(heartbeats.isEmpty());
+ }
+
+ @Test
+ public void testOnlineSubscriptionRefreshesGracePeriod() throws Exception {
+ List<HeartbeatData> heartbeats = recordHeartbeats();
+ this.clusterTransactionService.addTransactionSubscription(ctx, GROUP,
TOPIC);
+ expireSubscriptions();
+
+ when(this.producerManager.groupOnline(GROUP)).thenReturn(true);
+ this.clusterTransactionService.scanProducerHeartBeat();
+ assertEquals(1, heartbeats.size());
+
+ when(this.producerManager.groupOnline(GROUP)).thenReturn(false);
+ this.clusterTransactionService.scanProducerHeartBeat();
+
assertTrue(this.clusterTransactionService.getGroupClusterData().containsKey(GROUP));
+ assertEquals(1, heartbeats.size());
+
+ expireSubscriptions();
+ this.clusterTransactionService.scanProducerHeartBeat();
+
assertFalse(this.clusterTransactionService.getGroupClusterData().containsKey(GROUP));
+ assertEquals(1, heartbeats.size());
+ }
+
+ @Test
+ public void testReplacingSubscriptionRefreshesGracePeriod() throws
Exception {
+ List<HeartbeatData> heartbeats = recordHeartbeats();
+ this.clusterTransactionService.addTransactionSubscription(ctx, GROUP,
TOPIC);
+ expireSubscriptions();
+
+ this.clusterTransactionService.replaceTransactionSubscription(ctx,
GROUP, Lists.newArrayList(TOPIC));
+ this.clusterTransactionService.scanProducerHeartBeat();
+
+
assertTrue(this.clusterTransactionService.getGroupClusterData().containsKey(GROUP));
+ assertTrue(heartbeats.isEmpty());
+ }
+
+ @Test
+ public void
testUnsubscribeDuringGracePeriodRemovesSubscriptionImmediately() {
+ List<HeartbeatData> heartbeats = recordHeartbeats();
+ this.clusterTransactionService.addTransactionSubscription(ctx, GROUP,
TOPIC);
+ this.clusterTransactionService.scanProducerHeartBeat();
+
+ this.clusterTransactionService.unSubscribeAllTransactionTopic(ctx,
GROUP);
+
assertFalse(this.clusterTransactionService.getGroupClusterData().containsKey(GROUP));
+
+ when(this.producerManager.groupOnline(GROUP)).thenReturn(true);
+ this.clusterTransactionService.scanProducerHeartBeat();
+ assertTrue(heartbeats.isEmpty());
+ }
+
+ private List<HeartbeatData> recordHeartbeats() {
+ List<HeartbeatData> heartbeats = new ArrayList<>();
+ this.clusterTransactionService = new
ClusterTransactionService(this.topicRouteService, this.producerManager,
+ this.mqClientAPIFactory) {
+ @Override
+ protected void sendHeartBeatToCluster(String clusterName,
List<HeartbeatData> heartbeatDataList,
+ Map<String, String> brokerAddrNameMap) {
+ heartbeats.addAll(heartbeatDataList);
+ }
+ };
+ return heartbeats;
+ }
+
+ private void expireSubscriptions() throws Exception {
+ long expiredAt = System.currentTimeMillis() -
ConfigurationManager.getProxyConfig().getChannelExpiredTimeout()
+ - TimeUnit.SECONDS.toMillis(1);
+ Field lastActiveTimestamp =
ClusterTransactionService.ClusterData.class.getDeclaredField("lastActiveTimestamp");
+ lastActiveTimestamp.setAccessible(true);
+ for (ClusterTransactionService.ClusterData data :
this.clusterTransactionService.getGroupClusterData().get(GROUP)) {
+ lastActiveTimestamp.setLong(data, expiredAt);
+ }
+ }
+
@Test
public void testScanProducerHeartBeat() throws Exception {
when(this.producerManager.groupOnline(anyString())).thenReturn(true);
