ApsaraMQ for RocketMQ:Transactional messages

In an e-commerce scenario, when a user places an order, changes in the downstream systems are also triggered. For example, shipment is initiated in the logistics system, user credit points are updated in the points system, and the items are cleared in the shopping cart system. In this case, the following transaction branches are involved:

• Order system: The order status is changed from unpaid to paid.

• Logistics system: A to-be-shipped record is added, and a shipment record is created.

• Points system: The user credit points are updated.

• Shopping cart system: Items are cleared, and user records are updated.

However, this solution may deliver inconsistent results between the core transaction and transaction branches. Examples:

• The message is sent, but the order is not completed. In this case, the whole transaction must be rolled back.

• The order is completed, but the message is not sent. In this case, the message must be sent for consumption again.

• Timeout errors cannot be reliably detected, which makes it difficult to determine whether the order needs to be rolled back or an order change needs to be committed.

The following figure shows the process to use transactional messages.

1. The producer sends a message to an ApsaraMQ for RocketMQ broker.

2. The ApsaraMQ for RocketMQ broker persists the message and returns an acknowledgement (ACK) for the message to the producer. In this case, the message is marked as "not ready for delivery". A message in this state is called a half message.

3. The producer executes the local transaction.

4. The producer commits the execution result of the local transaction to the broker. The execution result can be Commit or Rollback. The following items describe the processing logic after the broker obtains the execution result:

   • If the execution result is Commit, the broker marks the half message as "ready for delivery" and delivers the message to the consumer.

   • If the execution result is Rollback, the broker rolls back the transaction and does not deliver the half message to the consumer.

5. If the broker does not receive the execution result or the status of the half message is Unknown due to a network disconnection or producer restart, the broker waits a period of time and sends a request to a producer in the producer cluster to query the status of the half message.

   Note

   For information about the interval between two status queries and the maximum number of queries, see Limits on parameters.

6. After the producer receives the request to query the status of the half message, the producer checks the execution result of the local transaction that corresponds to the half message.

7. The producer commits the execution result to the broker based on the queried status of the local transaction. Then, the broker performs the operations in Step 4 to process the half message.

Sending transactional messages is different from sending normal messages in the following aspects:

• Before you send transactional messages, you must enable the transaction checker and associate it with local transaction execution.

• To ensure transaction consistency, you must configure the transaction checker and bind the topics to which you want to send messages when you build the producer. If exceptions occur in the bound topics, the built-in transaction checker can be used to restore the status.

The following sample code provides an example on how to use transactional messages in Java.

For information about the complete sample code for messaging, see Apache RocketMQ 5.x SDKs (recommended).

    // The demo that is used to simulate the order table query service to check whether the order transaction is committed. 
    private static boolean checkOrderById(String orderId) {
        return true;
    }

    // The demo that is used to simulate the execution result of the local transaction. 
    private static boolean doLocalTransaction() {
        return true;
    }

    public static void main(String[] args) throws ClientException {
        ClientServiceProvider provider = new ClientServiceProvider();
        MessageBuilder messageBuilder = new MessageBuilder();
        // Build the transaction producer. The producer must build a transaction checker to check the intermediate status of an exceptional half message. 
        Producer producer = provider.newProducerBuilder()
                .setTransactionChecker(messageView -> {
                    /**
                     * The transaction checker checks whether the local transaction is correctly committed or rolled back based on the business ID. In this example, an order ID is used. 
                     * If this order is found in the order table, the order insertion action is correctly committed by the local transaction. If no order is found in the order table, the local transaction is rolled back. 
                     */
                    final String orderId = messageView.getProperties().get("OrderId");
                    if (Strings.isNullOrEmpty(orderId)) {
                        // An error occurs in the message. Rollback is returned. 
                        return TransactionResolution.ROLLBACK;
                    }
                    return checkOrderById(orderId) ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;
                })
                .build();
        // Create a transaction branch. 
        final Transaction transaction;
        try {
            transaction = producer.beginTransaction();
        } catch (ClientException e) {
            e.printStackTrace();
            // If the transaction branch fails to be created, the transaction is terminated. 
            return;
        }
        Message message = messageBuilder.setTopic("topic")
                // The message key. The system can use the key to locate the message. 
                .setKeys("messageKey")
                // The message tag. The consumer can use the tag to filter messages. 
                .setTag("messageTag")
                // The unique ID that is associated with the local transaction. The ID is used to verify the query of the local transaction status. 
                .addProperty("OrderId", "xxx")
                // The message body. 
                .setBody("messageBody".getBytes())
                .build();
        // Send the half message.
        final SendReceipt sendReceipt;
        try {
            sendReceipt = producer.send(message, transaction);
        } catch (ClientException e) {
            // If the half message fails to be sent, the transaction is terminated and the message is rolled back. 
            return;
        }
        /**
         * Execute the local transaction and determine the execution result. 
         * 1. If the result is Commit, deliver the message. 
         * 2. If the result is Rollback, roll back the message. 
         * 3. If an unknown exception occurs, no action is performed until a response is obtained from the query of the local transaction status. 
         *
         */
        boolean localTransactionOk = doLocalTransaction();
        if (localTransactionOk) {
            try {
                transaction.commit();
            } catch (ClientException e) {
                // You can determine whether to retry the message based on your business requirements. If you do not want to retry the message, you can query the local transaction status to commit the transaction status. 
                e.printStackTrace();
            }
        } else {
            try {
                transaction.rollback();
            } catch (ClientException e) {
                // We recommend that you record the exception information. This way, if an exception occurs during the rollback of a message, you can query the local transaction status to commit the transaction status without retrying the message. 
                e.printStackTrace();
            }
        }
    }

In most cases, a transaction is ongoing because the transaction execution is slow and the query of the local transaction status is too soon. The following solutions are provided:

• Specify a larger value for the time to perform the first query of the local transaction status. This may cause a large delay for transactions that depend on the query result.

• Make the program correctly identify ongoing transactions.