[Bug] PopMessageProcessor maybe no release lock

[zonghaishang]

Before Creating the Bug Report

• [X] I found a bug, not just asking a question, which should be created in GitHub Discussions.

• [X] I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.

• [X] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

mac os

RocketMQ version

5.3+

JDK Version

jdk 11

Describe the Bug

开启 enablePopMessageThreshold 触发popInflightMessageThreshold 可能会导致锁不释放。

    private CompletableFuture<Long> popMsgFromQueue(String topic, String attemptId, boolean isRetry, GetMessageResult getMessageResult,
        PopMessageRequestHeader requestHeader, int queueId, long restNum, int reviveQid,
        Channel channel, long popTime, ExpressionMessageFilter messageFilter, StringBuilder startOffsetInfo,
        StringBuilder msgOffsetInfo, StringBuilder orderCountInfo) {
        String lockKey =
            topic + PopAckConstants.SPLIT + requestHeader.getConsumerGroup() + PopAckConstants.SPLIT + queueId;
        boolean isOrder = requestHeader.isOrder();
        long offset = getPopOffset(topic, requestHeader.getConsumerGroup(), queueId, requestHeader.getInitMode(),
            false, lockKey, false);
        CompletableFuture<Long> future = new CompletableFuture<>();
        if (!queueLockManager.tryLock(lockKey)) {
            restNum = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - offset + restNum;
            future.complete(restNum);
            return future;
        }

        //  这里的逻辑是已经获取锁了，如果这里 return future; 就会导致锁未释放
        //  TODO:
        //  应该把 future.whenComplete((result, throwable) -> queueLockManager.unLock(lockKey)); 提前
        // 或者 放到  .... 下一行

        if (isPopShouldStop(topic, requestHeader.getConsumerGroup(), queueId)) {
            POP_LOGGER.warn("Too much msgs unacked, then stop poping. topic={}, group={}, queueId={}", topic, requestHeader.getConsumerGroup(), queueId);
            restNum = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - offset + restNum;
            future.complete(restNum);
            return future;
        }

        try {
            future.whenComplete((result, throwable) -> queueLockManager.unLock(lockKey));
        //  ....

Steps to Reproduce

代码走读发现

What Did You Expect to See?

代码走读发现

What Did You See Instead?

代码走读发现

Additional Context

代码走读发现

[yuz10]

https://github.com/apache/rocketmq/issues/8601