[Bug] messages will permanently send to other broker when enable sendLatencyFault

[Tcytw]

Before Creating the Bug Report

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

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

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

Runtime platform environment

all OS

RocketMQ version

4.9.x & 5.1.x

JDK Version

JDK1.8

Describe the Bug

After found the topicPublishRouteInfo, producer will select queue.
If enable sendLatencyFault, messageQueue will be replaced by other broker and queueId without recovering.
until producer reboot or broker reboot;

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        if (this.sendLatencyFaultEnable) {
            try {
                int index = tpInfo.getSendWhichQueue().incrementAndGet();
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    int pos = index++ % tpInfo.getMessageQueueList().size();
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                    if (!StringUtils.equals(lastBrokerName, mq.getBrokerName()) && latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
                        return mq;
                    }
                }

                final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
                int writeQueueNums = tpInfo.getWriteQueueIdByBroker(notBestBroker);
                if (writeQueueNums > 0) {
                    final MessageQueue mq = tpInfo.selectOneMessageQueue();
                    if (notBestBroker != null) {
                        mq.setBrokerName(notBestBroker);
                        mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
                    }
                    return mq;
                } else {
                    latencyFaultTolerance.remove(notBestBroker);
                }
            } catch (Exception e) {
                log.error("Error occurred when selecting message queue", e);
            }

            return tpInfo.selectOneMessageQueue();
        }

        return tpInfo.selectOneMessageQueue(lastBrokerName);
    }

Steps to Reproduce

send message continously with debug on some code for a while.

send message selectOneMessageQueue will set the queue.

then tryToFindTopicPublishInfo will find the queue which is changed

What Did You Expect to See?

the queue changed only at this send message, next send is not influenced

What Did You See Instead?

the queue changed not only at this send message, and queue is changed in the local cache

Additional Context

when set the queueId, I think messagQueue will return with a clone object instead of real message queue in topicRouteInfo

[RongtongJin]

@Tcytw Good catch! Could you submit a pull reqeust to fix the issue?

[leizhiyuan]

#7716 @RongtongJin

[Tcytw]

@Tcytw Good catch! Could you submit a pull reqeust to fix the issue?

@RongtongJin Of course. I have submitted the pull request at #7783

[RongtongJin]

@Tcytw Good catch! Could you submit a pull reqeust to fix the issue?

@RongtongJin Of course. I have submitted the pull request at #7783

It seems that @leizhiyuan had already submitted the same PR, and since he first proposed it, I will merge his PR into trunk.

[Tcytw]

@Tcytw Good catch! Could you submit a pull reqeust to fix the issue?

@RongtongJin Of course. I have submitted the pull request at #7783

It seems that @leizhiyuan had already submitted the same PR, and since he first proposed it, I will merge his PR into trunk.

okay