From 947ad6dfc7beb07e0287d971abd304996b0af2cf Mon Sep 17 00:00:00 2001 From: Felix-Gong Date: Wed, 21 Oct 2026 09:09:37 +0000 Subject: [PATCH] [ISSUE #11481] fix(broker): do not abort the queue-mapping clean pass on a missing remote queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cleanItemListMoreThanSecondGen() validates, for every queue whose leader moved, that the new leader agrees. It reads the new leader's hosted queues with get(qId) and calls isEmpty() on the result unconditionally: when the remote broker's mapping does not contain that queue id at all — the transient state this service exists to reconcile — the read returns null and the pass dies with a NullPointerException. The exception is swallowed by the per-topic catch, so the visible damage is silent: every remaining queue of that topic is skipped for this cycle, the mapping is left half-cleaned, and the failure repeats every five minutes. Treat a missing remote queue as "not confirmed yet" and continue, which is how the surrounding reads already handle absent data. Red/green: remoteLeaderMissingTheQueueIdDoesNotAbortTheRestOfTheTopic fails on the old code (the second queue is never removed because the NPE aborts the pass) and passes with the null guard. Broker module: 815 tests, the 7 failures are pre-existing (RocksDB/filter/schedule/pop-lite/ BrokerOuterAPI) and reproduce identically with this change stashed. --- .../topic/TopicQueueMappingCleanService.java | 2 +- .../TopicQueueMappingCleanServiceTest.java | 59 +++++++++++++++++++ 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanService.java b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanService.java index c4fd4c62082..7a78ee91e6e 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanService.java @@ -294,7 +294,7 @@ public void cleanItemListMoreThanSecondGen() { continue; } List items = remoteMappingDetail.getHostedQueues().get(qId); - if (items.isEmpty()) { + if (items == null || items.isEmpty()) { continue; } LogicQueueMappingItem leaderItem = items.get(items.size() - 1); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanServiceTest.java index c7079c5248f..72cc1f6fe0c 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanServiceTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanServiceTest.java @@ -19,13 +19,18 @@ import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.out.BrokerOuterAPI; import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.TopicConfig; import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.remoting.exception.RemotingException; import org.apache.rocketmq.remoting.protocol.DataVersion; +import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.apache.rocketmq.remoting.protocol.admin.TopicOffset; import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable; +import org.apache.rocketmq.remoting.protocol.header.GetTopicConfigRequestHeader; import org.apache.rocketmq.remoting.protocol.route.TopicRouteData; import org.apache.rocketmq.remoting.protocol.statictopic.LogicQueueMappingItem; +import org.apache.rocketmq.remoting.protocol.statictopic.TopicConfigAndQueueMapping; +import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingInfo; import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail; import org.apache.rocketmq.remoting.rpc.RpcClient; import org.apache.rocketmq.remoting.rpc.RpcRequest; @@ -38,12 +43,15 @@ import org.mockito.junit.MockitoJUnitRunner; import java.util.Arrays; +import java.util.HashMap; import java.util.Collections; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyLong; @@ -166,4 +174,55 @@ public void testCleanItemListMoreThanSecondGenException() throws Exception { topicQueueMappingCleanService.cleanItemListMoreThanSecondGen(); verify(brokerOuterAPI, times(1)).getTopicRouteInfoFromNameServer(any(), anyLong()); } + + @Test + public void remoteLeaderMissingTheQueueIdDoesNotAbortTheRestOfTheTopic() throws Exception { + when(brokerConfig.getBrokerName()).thenReturn("brokerA"); + when(messageStoreConfig.getDeleteWhen()).thenReturn(deleteWhen); + + TopicQueueMappingDetail localDetail = new TopicQueueMappingDetail(defaultTopic, 2, "brokerA", 1); + localDetail.getHostedQueues().put(0, Arrays.asList( + new LogicQueueMappingItem(0, 0, "brokerOld", 0, 0, 100, 0, 0), + new LogicQueueMappingItem(1, 0, "brokerB", 100, 100, 200, 0, 0))); + localDetail.getHostedQueues().put(1, Arrays.asList( + new LogicQueueMappingItem(0, 1, "brokerOld", 0, 0, 100, 0, 0), + new LogicQueueMappingItem(1, 1, "brokerB", 100, 100, 200, 0, 0))); + when(topicQueueMappingManager.getTopicQueueMappingTable()) + .thenReturn(new ConcurrentHashMap<>(Collections.singletonMap(defaultTopic, localDetail))); + + TopicRouteData routeData = new TopicRouteData(); + Map mappingByBroker = new HashMap<>(); + TopicQueueMappingInfo infoB = new TopicQueueMappingInfo(defaultTopic, 2, "brokerB", 1); + infoB.getCurrIdMap().put(0, 0); + mappingByBroker.put("brokerB", infoB); + TopicQueueMappingInfo infoC = new TopicQueueMappingInfo(defaultTopic, 2, "brokerC", 1); + infoC.getCurrIdMap().put(1, 1); + mappingByBroker.put("brokerC", infoC); + routeData.setTopicQueueMappingByBroker(mappingByBroker); + when(brokerOuterAPI.getTopicRouteInfoFromNameServer(any(), anyLong())).thenReturn(routeData); + + // brokerB no longer hosts qid 0 at all, brokerC confirms it owns qid 1 + TopicQueueMappingDetail remoteB = new TopicQueueMappingDetail(defaultTopic, 2, "brokerB", 1); + TopicQueueMappingDetail remoteC = new TopicQueueMappingDetail(defaultTopic, 2, "brokerC", 1); + remoteC.getHostedQueues().put(1, Arrays.asList( + new LogicQueueMappingItem(0, 1, "brokerOld", 0, 0, 100, 0, 0), + new LogicQueueMappingItem(1, 1, "brokerC", 100, 100, 200, 0, 0))); + when(rpcClient.invoke(any(RpcRequest.class), anyLong())).thenAnswer(invocation -> { + RpcRequest rpcRequest = invocation.getArgument(0); + String broker = ((GetTopicConfigRequestHeader) rpcRequest.getHeader()).getBrokerName(); + TopicQueueMappingDetail remoteDetail = "brokerC".equals(broker) ? remoteC : remoteB; + return CompletableFuture.completedFuture(new RpcResponse(ResponseCode.SUCCESS, null, + new TopicConfigAndQueueMapping(new TopicConfig(defaultTopic, 2, 2), remoteDetail))); + }); + DataVersion dataVersion = mock(DataVersion.class); + when(topicQueueMappingManager.getDataVersion()).thenReturn(dataVersion); + + topicQueueMappingCleanService.cleanItemListMoreThanSecondGen(); + + assertFalse("a queue whose leader moved away must be cleaned even though another queue " + + "of the same topic is missing from its remote mapping", + localDetail.getHostedQueues().containsKey(1)); + assertTrue("a queue that is still led by this broker's item list stays", + localDetail.getHostedQueues().containsKey(0)); + } }