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

window

RocketMQ version

branch: release-4.4.0 Even earlier ~ develop

JDK Version

1.8

Describe the Bug

在阅读源码时发现的问题：

1. bug1: broker 端最后一次检测 half 消息时，没有将应该拉取的所有 op 消息拉取到内存中，就判断 half 对应的 op 消息存不在不存在，可能错误地认为不存在。导致该 half 丢弃，冗余地进入了 "TRANS_CHECK_MAX_TIME_TOPIC"。
2. bug2: 生产的事务消息配置了长的 "PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS"，若生产端的 check 一直返回 "LocalTransactionState.UNKNOW"，broker 端发送 check 请求的次数不会达到应该要的 "transactionCheckMax" 次，极端情况下，一次 check 请求都不会发

Steps to Reproduce

bug2:

1. 发送 "PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS" = 600 的事务消息，主动发送 LocalTransactionState.UNKNOW，期望回查。
2. broker 端间隔 30 秒 check 一次这条半消息，每次先对该半消息的回查次数+1，然后发现在 IMMUNITY 期间内，重新投递回查次数+1的消息，不真正发起回查请求。
3. 30 秒 * 15 次后，半消息的回查次数到达 "transactionCheckMax" 次，消息被丢弃到 "TRANS_CHECK_MAX_TIME_TOPIC"。整个期间未发起过回查请求。

    public static void main(String[] args) throws Exception {
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("producerGroup");
        transactionMQProducer.setNamesrvAddr("127.0.0.1:9876");
        transactionMQProducer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                return LocalTransactionState.UNKNOW;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // CANNOT EXECUTE !
                return null;
            }
        });
        transactionMQProducer.start();
        Message msg = new Message();
        msg.setTopic("topic");
        msg.setBody("body".getBytes());
        msg.putUserProperty(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS, "600");
        transactionMQProducer.sendMessageInTransaction(msg, null);
        Thread.sleep(1000 * 1000 * 1000);
    }

What Did You Expect to See?

TransactionalMessageServiceImpl adjust needDiscard(...) position

What Did You See Instead?

TransactionalMessageServiceImpl needDiscard(...) position error

Additional Context

No response