ApsaraMQ for RocketMQ:Send and receive transactional messages

Step 1: Implement the transaction check listener

The broker calls this listener when it needs to verify the outcome of a local transaction. After a transactional message is sent, ApsaraMQ for RocketMQ calls the LocalTransactionChecker API operation to respond to the status check request from the broker. Your implementation should query the actual transaction state -- for example, by checking whether an order exists in your database -- and return the appropriate status.

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.common.message.MessageExt;

public class LocalTransactionCheckerImpl implements TransactionCheckListener {

    @Override
    public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
        System.out.println("Received transaction status check request. MsgId: " + msg.getMsgId());

        // Query your business data to determine the transaction outcome.
        // Example: Check whether the order associated with this message
        // was committed to the database.
        //
        // boolean orderExists = orderService.existsByOrderId(msg.getKeys());
        // if (orderExists) {
        //     return LocalTransactionState.COMMIT_MESSAGE;
        // }
        // return LocalTransactionState.ROLLBACK_MESSAGE;

        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

Step 2: Send a half message and execute the local transaction

The producer sends a half message to the broker and immediately executes the local transaction in the executeLocalTransactionBranch callback. Based on the return value, the broker commits, rolls back, or schedules a status check for the message.

Replace the following placeholders with your actual values:

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class RocketMQTransactionProducer {

    // Load AccessKey credentials from environment variables.
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a transactional producer with message trace enabled.
        // To disable message trace, use:
        // new TransactionMQProducer("YOUR TRANSACTION GROUP ID", getAclRPCHook());
        TransactionMQProducer producer = new TransactionMQProducer(
            null, "YOUR TRANSACTION GROUP ID", getAclRPCHook(), true, null);

        // Set the endpoint from the ApsaraMQ for RocketMQ console.
        producer.setNamesrvAddr("http://xxxx.mq-internet.aliyuncs.com:80");

        // Required for message trace on Alibaba Cloud.
        producer.setAccessChannel(AccessChannel.CLOUD);

        // Register the transaction check listener (see Step 1).
        producer.setTransactionCheckListener(new LocalTransactionCheckerImpl());

        producer.start();

        for (int i = 0; i < 10; i++) {
            try {
                Message message = new Message(
                    "YOUR TRANSACTION TOPIC",
                    "YOUR MESSAGE TAG",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

                // Send the half message and execute the local transaction.
                SendResult sendResult = producer.sendMessageInTransaction(
                    message,
                    new LocalTransactionExecuter() {
                        @Override
                        public LocalTransactionState executeLocalTransactionBranch(
                                Message msg, Object arg) {
                            System.out.println("Executing local transaction: " + msg);
                            // Run your business logic here (e.g., insert order into DB).
                            // Return COMMIT_MESSAGE, ROLLBACK_MESSAGE, or UNKNOW
                            // based on the outcome.
                            return LocalTransactionState.UNKNOW;
                        }
                    },
                    null);
                assert sendResult != null;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

Transaction states

Why the transaction status check matters

If the producer returns UNKNOW, crashes, or loses connectivity after sending the half message, the broker cannot determine whether to deliver or discard the message. The transaction status check resolves this by periodically querying the producer until it receives a definitive commit or rollback.

Your LocalTransactionCheckerImpl must:

1. Query the state of the local transaction that corresponds to the half message -- for example, look up the order in your database.

2. Return the correct LocalTransactionState to the broker.