From d7821670091e1512a35ff62368f80374c5ccaa2b Mon Sep 17 00:00:00 2001 From: zjncs <18910855655@163.com> Date: Wed, 9 Sep 2026 18:02:45 +0800 Subject: [PATCH] [ISSUE #B3] Read the consume type from the compensation table in the cold-data flow-control path The cold-data flow-control branch of PullMessageProcessor read the consume type with ConsumerManager.getConsumerGroupInfo(group), which only sees the live consumer table. Pull requests that carry their subscription (proxy / lite-pull traffic) never registered the group there, so once the group is flagged for cold-data flow control and the pulled offset is cold, the processor dereferenced null and failed the request with an NPE. The group's basic info (consume type, message model) is already compensated into consumerCompensationTable a few lines earlier for exactly these pull styles, so read it with getConsumerGroupInfo(group, true) like ConsumerLagCalculator does, and fall back to the passive flow-control response if even the compensation table has no record. Signed-off-by: zjncs <18910855655@163.com> --- .../processor/PullMessageProcessor.java | 9 ++++- .../processor/PullMessageProcessorTest.java | 40 +++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java index d8e026a16b0..b2933145450 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java @@ -513,7 +513,14 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re .getColdDataCheckService().isMsgInColdArea(requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getQueueOffset()); if (isMsgLogicCold) { - ConsumeType consumeType = this.brokerController.getConsumerManager().getConsumerGroupInfo(requestHeader.getConsumerGroup()).getConsumeType(); + // The group may only be known through the compensation table filled a + // few lines above (pull consumers and proxy traffic), so look there + // too; if even that has no record, flow control like a passive + // consumer instead of failing the request with an NPE. + ConsumerGroupInfo consumerGroupInfo = this.brokerController.getConsumerManager() + .getConsumerGroupInfo(requestHeader.getConsumerGroup(), true); + ConsumeType consumeType = consumerGroupInfo == null + ? ConsumeType.CONSUME_PASSIVELY : consumerGroupInfo.getConsumeType(); if (consumeType == ConsumeType.CONSUME_PASSIVELY) { response.setCode(ResponseCode.SYSTEM_BUSY); response.setRemark("This consumer group is reading cold data. It has been flow control"); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java index cecd1ff86a9..c52486b10f8 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java @@ -35,6 +35,7 @@ import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.TopicConfig; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +import org.apache.rocketmq.common.sysflag.PullSysFlag; import org.apache.rocketmq.remoting.exception.RemotingCommandException; import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.apache.rocketmq.remoting.netty.NettyServerConfig; @@ -49,6 +50,8 @@ import org.apache.rocketmq.store.GetMessageResult; import org.apache.rocketmq.store.GetMessageStatus; import org.apache.rocketmq.store.MessageStore; +import org.apache.rocketmq.store.CommitLog; +import org.apache.rocketmq.store.DefaultMessageStore; import org.apache.rocketmq.store.config.MessageStoreConfig; import org.junit.Assert; import org.junit.Before; @@ -63,6 +66,7 @@ import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @RunWith(MockitoJUnitRunner.class) @@ -247,6 +251,42 @@ public void testCommitPullOffset() throws RemotingCommandException { .isEqualTo(getMessageResult.getNextBeginOffset()); } + @Test + public void testColdDataFlowCtrWhenGroupIsOnlyInCompensationTable() throws Exception { + brokerController.getMessageStoreConfig().setColdDataFlowControlEnable(true); + brokerController.getColdDataCgCtrService().coldAcc(group, Long.MAX_VALUE); + brokerController.getConsumerManager().unregisterConsumer(group, clientChannelInfo, false); + + DefaultMessageStore defaultMessageStore = mock(DefaultMessageStore.class); + CommitLog commitLog = mock(CommitLog.class); + CommitLog.ColdDataCheckService coldDataCheckService = mock(CommitLog.ColdDataCheckService.class); + when(defaultMessageStore.getCommitLog()).thenReturn(commitLog); + when(commitLog.getColdDataCheckService()).thenReturn(coldDataCheckService); + when(coldDataCheckService.isMsgInColdArea(anyString(), anyString(), anyInt(), anyLong())).thenReturn(true); + brokerController.setMessageStore(defaultMessageStore); + + // Pull with the subscription carried in the request (proxy / lite-pull style): + // the group is absent from the live consumer table, only compensated above. + PullMessageRequestHeader requestHeader = new PullMessageRequestHeader(); + requestHeader.setCommitOffset(123L); + requestHeader.setConsumerGroup(group); + requestHeader.setMaxMsgNums(100); + requestHeader.setQueueId(1); + requestHeader.setQueueOffset(456L); + requestHeader.setSubscription("*"); + requestHeader.setTopic(topic); + requestHeader.setSysFlag(PullSysFlag.buildSysFlag(false, false, true, false)); + requestHeader.setSubVersion(100L); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, requestHeader); + request.makeCustomHeaderToNet(); + + RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request); + + assertThat(response).isNotNull(); + assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_BUSY); + assertThat(response.getRemark()).contains("cold data"); + } + private RemotingCommand createPullMsgCommand(int requestCode) { PullMessageRequestHeader requestHeader = new PullMessageRequestHeader(); requestHeader.setCommitOffset(123L);