[BUG] Cold-data flow control NPEs for pull requests whose group is only in the compensation table

[zjncs]

Motivation

When a broker has coldDataFlowControlEnable=true and a consumer group is flagged for cold-data flow control (ColdDataCgCtrService), the cold branch of PullMessageProcessor does:

ConsumeType consumeType = this.brokerController.getConsumerManager()
    .getConsumerGroupInfo(requestHeader.getConsumerGroup()).getConsumeType();

getConsumerGroupInfo(group) only consults the live consumer table. But pull requests that carry their subscription in the request (proxy / lite-pull traffic, i.e. hasSubscriptionFlag=true\)) never registered the group there — the broker instead compensates the group's basic info into consumerCompensationTable a few lines earlier (compensateBasicConsumerInfo, PullMessageProcessor` L381-392). For those requests the lookup returns null and the pull thread dies with:

java.lang.NullPointerException: Cannot invoke "ConsumerGroupInfo.getConsumeType()"
  because the return value of "ConsumerManager.getConsumerGroupInfo(String)" is null

So a flow-controlled group reading cold data via the proxy gets NPE responses instead of the intended flow control.

Changes

• Look the group up with getConsumerGroupInfo(group, true) so the compensation table filled a few lines above is also consulted (same pattern as ConsumerLagCalculator).
• If even the compensation table has no record, fall back to CONSUME_PASSIVELY — the conservative flow-control response — instead of dereferencing null.

Verification

New test PullMessageProcessorTest#testColdDataFlowCtrWhenGroupIsOnlyInCompensationTable: enables cold-data flow control, flags the group via coldAcc, unregisters the live consumer, mocks a DefaultMessageStore whose commit log reports the offset as cold, and sends a subscription-carrying pull request.

$ mvn -pl broker test -Dtest='PullMessageProcessorTest'
(before) java.lang.NullPointerException: Cannot invoke ... getConsumeType() ... is null
(after)  Tests run: 11, Failures: 0, Errors: 0   (new test asserts SYSTEM_BUSY + cold-data remark)

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[RockteMQ-AI]

Summary

Uses getConsumerGroupInfo(group, true) to also consult the compensation table when looking up consumer group info for cold-data flow control — prevents NPE when the group exists only in the compensation table. Falls back to CONSUME_PASSIVELY if the group is not found anywhere, which is the safe default. Well-documented with a thorough regression test.

LGTM.

---

Automated review by github-manager-bot