Transactional Message Sending

Introduction

In some scenarios where there is a strong need for data consistency, Apache RocketMQ transactional messages can be used to ensure consistency of upstream and downstream data.

事务消息1

Transactional messages are send in two phases. At first, a half message will be delivered, which refers to a message is successfully sent to the MQ server, but the server did not receive the second acknowledgement of the message from the Producer, then the message will be marked as “temporarily undeliverable” state.

The local transaction will be executed if the message is sent successfully, and a half message status (commit or rollback) will be delivered to the Broker depending on the local transaction results.

If the second acknowledgement of a transactional message is lost due to network flashback, Producer restart, etc., the Broker will find the message which is in “half message” state for a long time, and take the initiative to check the transaction status of the message (Commit or Rollback) from the Producer. Therefore, the downstream will receive the message if the local transaction is executed successfully, otherwise it will not. This ultimately ensures the consistency of the upstream and downstream data.

The detailed execute flow of the transactional message is shown in the following diagram:

事务消息2

Transactional Message Sending Procedure

  1. The Producer sends the half message to the RocketMQ Broker.

  2. After the RocketMQ Broker persists the message successfully, it returns an Ack to the Producer confirming that the message was sent successfully and it is a half message.

  3. The Producer starts executing the local transaction.

  4. The Producer submits a second acknowledgement (Commit or Rollback) to the server based on the result of the local transaction, and the server receives the acknowledgment and processes the logic as follows.

    • If the second acknowledgement result is Commit: the server marks the half message as deliverable and delivers it to the Consumer.
    • If the second acknowledgement result is Rollback: the server will rollback the transaction and will not deliver the half message to the Consumer.
  5. In the special case of network disconnection or the Producer restarts, if the server does not receive the second acknowledgment result from the Producer, or the second acknowledgment result received by the server is Unknown, the server will initiate a rollback message to a Producer after a fixed time.

The procedure of the transaction status check are as follows.

  1. After receiving the transaction status check request, the Producer needs to verify the final result of the local transaction of the corresponding message.
  2. The producer submits the second acknowledgment again based on the final result of the local transaction, and the server side will still processes the half message according to step 4.

Example

  1. public class TransactionProducer {
  2.     public static void main(String[] args) throws MQClientException, InterruptedException {
  3.         TransactionListener transactionListener = new TransactionListenerImpl();
  4.         TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
  5.         ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
  6.             @Override
  7.             public Thread newThread(Runnable r) {
  8.                 Thread thread = new Thread(r);
  9.                 thread.setName("client-transaction-msg-check-thread");
  10.                 return thread;
  11.             }
  12.         });
  14.         producer.setExecutorService(executorService);
  15.         producer.setTransactionListener(transactionListener);
  16.         producer.start();
  18.         String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
  19.         for (int i = 0; i < 10; i++) {
  20.             try {
  21.                 Message msg =
  22.                     new Message("TopicTest", tags[i % tags.length], "KEY" + i,
  23.                         ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
  24.                 SendResult sendResult = producer.sendMessageInTransaction(msg, null);
  25.                 System.out.printf("%s%n", sendResult);
  27.                 Thread.sleep(10);
  28.             } catch (MQClientException | UnsupportedEncodingException e) {
  29.                 e.printStackTrace();
  30.             }
  31.         }
  33.         for (int i = 0; i < 100000; i++) {
  34.             Thread.sleep(1000);
  35.         }
  36.         producer.shutdown();
  37.     }
  39.     static class TransactionListenerImpl implements TransactionListener {
  40.         private AtomicInteger transactionIndex = new AtomicInteger(0);
  42.         private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
  44.         @Override
  45.         public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
  46.             int value = transactionIndex.getAndIncrement();
  47.             int status = value % 3;
  48.             localTrans.put(msg.getTransactionId(), status);
  49.             return LocalTransactionState.UNKNOW;
  50.         }
  52.         @Override
  53.         public LocalTransactionState checkLocalTransaction(MessageExt msg) {
  54.             Integer status = localTrans.get(msg.getTransactionId());
  55.             if (null != status) {
  56.                 switch (status) {
  57.                     case 0:
  58.                         return LocalTransactionState.UNKNOW;
  59.                     case 1:
  60.                         return LocalTransactionState.COMMIT_MESSAGE;
  61.                     case 2:
  62.                         return LocalTransactionState.ROLLBACK_MESSAGE;
  63.                     default:
  64.                         return LocalTransactionState.COMMIT_MESSAGE;
  65.                 }
  66.             }
  67.             return LocalTransactionState.COMMIT_MESSAGE;
  68.         }
  69.     }
  70. }

Transactional messages are no longer sent by DefaultMQProducer, but using TransactionMQProducer. The above sample sets the thread pool for the transactional message check, if not, one will be generated by default. The most important thing is to implement the TransactionListener interface and pass TransactionMQProducer into it.

Transactional Message Sending - 图3note

The TransactionListener interface is defined as follows:

  1. public interface TransactionListener {
  2.     /**
  3.      * When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.
  4.      *
  5.      * @param msg Half(prepare) message
  6.      * @param arg Custom business parameter
  7.      * @return Transaction state
  8.      */
  9.     LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);
  11.     /**
  12.      * When no response to prepare(half) message. broker will send check message to check the transaction status, and this
  13.      * method will be invoked to get local transaction status.
  14.      *
  15.      * @param msg Check message
  16.      * @return Transaction state
  17.      */
  18.     LocalTransactionState checkLocalTransaction(final MessageExt msg);
  19. }

executeLocalTransaction is the method that executes the local transaction after the half message has been sent successfully. After executing the local transaction, the following three states can be returned in this method.

  • LocalTransactionState.COMMIT_MESSAGE: the transaction is committed, allowing the consumer to consume the message.
  • LocalTransactionState.ROLLBACK_MESSAGE: the transaction is rolled back, and the message will be discarded without being allowed to be consumed.
  • LocalTransactionState.UNKNOW: temporarily unable to determine the state. After waiting for a fixed time, the Broker send the transaction status check message back to the producer.

checkLocalTransaction is a method to check the transaction state on the Broker side because the second acknowledgement is not received. Transaction status check rule: After the execution of the local transaction is completed, if the local transaction returns LocalTransactionState.UNKNOW status to the Broker, or the Producer exits causing no status returned from the Producer. Then the Broker will initiate a transaction status check message to the Producer, and it will check again at regular intervals if the transaction status is still not obtained.

Transactional Message Sending - 图4caution

It is important to note that the ProducerGroupName of a transactional message cannot be set arbitrarily. Transactional messages have a transaction status check mechanism. If the original Producer is found to have crashed and collapsed, the Broker will contact other Producer instances within the same Producer group to check the local transaction execution and Commit or Rollback half messages.