From 856d25d26e0b26ad22829b60831973cc18d28ebc Mon Sep 17 00:00:00 2001 From: liuhy Date: Sat, 15 Aug 2026 01:23:21 -0700 Subject: [PATCH] fix(proxy): initialize cluster transaction address map Signed-off-by: liuhy --- .../service/transaction/ClusterTransactionService.java | 3 ++- .../transaction/ClusterTransactionServiceTest.java | 10 ++++++++++ 2 files changed, 12 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 1ec42864636..6bf731d5574 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 @@ -57,7 +57,8 @@ public class ClusterTransactionService extends AbstractTransactionService { private ThreadPoolExecutor heartbeatExecutors; private final Map/* cluster list */> groupClusterData = new ConcurrentHashMap<>(); - private final AtomicReference> brokerAddrNameMapRef = new AtomicReference<>(); + private final AtomicReference> brokerAddrNameMapRef = + new AtomicReference<>(Collections.emptyMap()); private TxHeartbeatServiceThread txHeartbeatServiceThread; public ClusterTransactionService(TopicRouteService topicRouteService, ProducerManager producerManager, 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 91af74cbefc..a5d459e90c5 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 @@ -26,6 +26,7 @@ import java.util.stream.Collectors; import org.apache.rocketmq.broker.client.ProducerManager; import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.service.BaseServiceTest; @@ -44,6 +45,7 @@ import static org.awaitility.Awaitility.await; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; @@ -108,6 +110,14 @@ public void testUnSubscribeAllTransactionTopic() { assertEquals(0, this.clusterTransactionService.getGroupClusterData().size()); } + @Test + public void testAddTransactionDataByBrokerAddrBeforeFirstHeartbeat() { + TransactionData transactionData = this.clusterTransactionService.addTransactionDataByBrokerAddr( + ctx, BROKER_ADDR, TOPIC, GROUP, 1, 2, "transaction-id", new Message()); + + assertNull(transactionData); + } + @Test public void testScanProducerHeartBeat() throws Exception { when(this.producerManager.groupOnline(anyString())).thenReturn(true);