Sending and Receiving Transactional Messages

DMS for RocketMQ ensures transaction consistency between the service logic and message transmission, and implements transaction support in two phases. Figure 1 illustrates the interaction of transactional messages.

Figure 1 Transactional message interaction

The producer sends a half message and then executes the local transaction. If the execution is successful, the transaction is committed. If the execution fails, the transaction is rolled back. If the server does not receive any commit or rollback request after a period of time, it initiates a check. After receiving the check request, the producer resends a transaction commit or rollback request. The message is delivered to the consumer only after being committed. The consumer is unaware of the rollback.

Before sending and receiving transactional messages, collect RocketMQ connection information by referring to Collecting Connection Information.

Preparing the Environment

You can connect open-source Java clients to DMS for RocketMQ. The recommended Java client version is 4.9.8.

Use either of the following methods to import a dependency:

Using Maven

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.8</version>
</dependency>

Downloading the dependency.

Sending Transactional Messages

Refer to the following sample code or obtain more sample code from TransactionProducer.java.

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;

public class Main {
    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException {
        TransactionListener transactionListener = new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("Start to execute the local transaction: "+ message);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("Check request received. Check the transaction status: "+ messageExt);
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        };

        TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
        // Enter the address.
        producer.setNamesrvAddr("192.168.0.1:8100");
        //producer.setUseTLS(true);    // Add this line if SSL has been enabled during instance creation.
        producer.setTransactionListener(transactionListener);
        producer.start();

        Message msg =
            new Message("TopicTest", "TagA", "KEY",
                "Hello RocketMQ ".getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.sendMessageInTransaction(msg, null);
        System.out.printf("%s%n", sendResult);

        producer.shutdown();
    }}

The producer needs to implement two callback functions. The executeLocalTransaction callback function is called after the half message is sent (see step 3 in the diagram). The checkLocalTransaction callback function is called after the check request is received (see step 6 in the diagram). The two callback functions can return three transaction states:

LocalTransactionState.COMMIT_MESSAGE: Transaction committed. The consumer can retrieve the message.
LocalTransactionState.ROLLBACK_MESSAGE: Transaction rolled back. The message will be discarded and cannot be retrieved.
LocalTransactionState.UNKNOW: The status cannot be determined and the server is expected to check the message status from the producer again.