事务消息提交后依然会出现回查

[lizhimins]

事务消息成功提交后依然会出现回查

分支 develop 最新，broker Config 设置 transactionCheckInterval=6000

为了更好的复现这个问题，我提取了回查部分的实现，如下：

public class TransactionProducer {

    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionListener transactionListener = new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.out.println("no execute local trans" + msg.getTransactionId());
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.printf("%s return rollback %s%n",
                    new Date(System.currentTimeMillis()), msg.getTransactionId());
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        };

        String groupId = "PID-trans";
        TransactionMQProducer producer = new TransactionMQProducer(groupId);
        producer.setNamesrvAddr("11.165.57.3:9876");
        producer.setVipChannelEnabled(false);
        ExecutorService executorService = new ThreadPoolExecutor(
            1, 1, 100, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(2000), r -> {
            Thread thread = new Thread(r);
            thread.setName("client-transaction");
            return thread;
        });

        producer.setExecutorService(executorService);
        producer.setTransactionListener(transactionListener);
        producer.start();

        for (int i = 0; i < 10; i++) {
            try {
                Message msg =
                    new Message("flink-sink-1", "tag", "key" + i,
                        ("Hello World" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

                MessageAccessor.putProperty(msg,
                    MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS, String.valueOf(30));
                MessageAccessor.putProperty(
                    msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, Boolean.TRUE.toString().toLowerCase());
                MessageAccessor.putProperty(
                    msg, MessageConst.PROPERTY_PRODUCER_GROUP, groupId);

                SendResult sendResult = producer.send(msg);

                if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
                    if (sendResult.getTransactionId() != null) {
                        msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
                    }
                    String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
                    if (null != transactionId && !"".equals(transactionId)) {
                        msg.setTransactionId(transactionId);
                    }
                }

                System.out.printf("%s send success %s%n", new Date(System.currentTimeMillis()), sendResult.getMsgId());

                TimeUnit.SECONDS.sleep(1);

                // noinspection deprecation
                producer.getDefaultMQProducerImpl().endTransaction(
                    msg, sendResult, LocalTransactionState.COMMIT_MESSAGE, null);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();
    }
}

end transaction 之后依然出现了无效回查

[lizhimins]

通过 queryMsgByUnikey 可以查询到，commit message 已经生效，消息正常进入了业务 topic

[bingfan1005]

问题解决了吗 是什么原因啊 ？ 我也遇见了

[asjfoajs]

System.out.printf("%s send success %s%n", new Date(System.currentTimeMillis()), sendResult.getMsgId()); 只是发送成功，// noinspection deprecation
producer.getDefaultMQProducerImpl().endTransaction(
msg, sendResult, LocalTransactionState.COMMIT_MESSAGE, null); 才是提交成功 不应该是在这个后面打个日志看一下是否在checkLocalTransaction之前执行吗？