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.
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.
Notes and Constraints
- The gRPC protocol is only supported by RocketMQ v5.x but not v4.8.0.
- To receive and send transactional messages, ensure the topic message type is Transactional before connecting a client to a RocketMQ instance of v5.x.
Preparing the Environment
You can connect open-source Java clients to DMS for RocketMQ. The recommended Java client version is 5.0.5.
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client-java</artifactId> <version>5.0.5</version> </dependency>
Sending Transactional Messages
Refer to the following sample code or obtain more sample code from ProducerTransactionMessageExample.java.
import org.apache.rocketmq.client.apis.ClientConfiguration; import org.apache.rocketmq.client.apis.ClientException; import org.apache.rocketmq.client.apis.ClientServiceProvider; import org.apache.rocketmq.client.apis.SessionCredentialsProvider; import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider; import org.apache.rocketmq.client.apis.message.Message; import org.apache.rocketmq.client.apis.producer.Producer; import org.apache.rocketmq.client.apis.producer.SendReceipt; import org.apache.rocketmq.client.apis.producer.Transaction; import org.apache.rocketmq.client.apis.producer.TransactionChecker; import org.apache.rocketmq.client.apis.producer.TransactionResolution; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.nio.charset.StandardCharsets; public class ProducerTransactionMessageExample { private static final Logger log = LoggerFactory.getLogger(ProducerTransactionMessageExample.class); public static void main(String[] args) throws ClientException, IOException { final ClientServiceProvider provider = ClientServiceProvider.loadService(); String topic = "yourTransactionTopic"; // Specify the gRPC address or gRPC public address String endpoints = "yourEndpoints"; // Enter your username and key. Hard-coded or plaintext username and key are risky. You are advised to store them in ciphertext in a configuration file or an environment variable. The code below is required only if ACL was enabled during instance creation. String accessKey = System.getenv("ROCKETMQ_AK"); String secretKey = System.getenv("ROCKETMQ_SK"); SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey); ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder() .setEndpoints(endpoints) // .enableSsl(false) // This line is mandatory to create an instance with SSL set to PLAINTEXT. This line is optional to create an instance with SSL set to PERMISSIVE. // .setCredentialProvider(sessionCredentialsProvider) // Set the credential provider if ACL has been enabled. .build(); TransactionChecker checker = messageView -> { log.info("Receive transactional message check, message={}", messageView); // Return the transaction resolution according to your business logic. // Check the local transaction and return its status. return TransactionResolution.COMMIT; }; final Producer producer = provider.newProducerBuilder() .setClientConfiguration(clientConfiguration) .setTopics(topic) // The producer needs to construct a transaction checker to check the intermediate status of abnormal transactions. .setTransactionChecker(checker) .build(); byte[] body = "This is a transaction message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8); String tag = "yourMessageTagA"; final Message message = provider.newMessageBuilder() .setTopic(topic) .setTag(tag) .setKeys("yourMessageKey") .setBody(body) .build(); // Start a transaction branch. final Transaction transaction; try { transaction = producer.beginTransaction(); } catch (ClientException e) { log.error("Failed to begin transaction", e); // The transaction branch fails to be started and exits. return; } try { final SendReceipt sendReceipt = producer.send(message, transaction); log.info("Send transaction message successfully, messageId={}", sendReceipt.getMessageId()); } catch (Throwable t) { log.error("Failed to send message", t); return; } /** * Execute a local transaction and check the result. * 1. Commit a transactional message if local transaction is committed. * 2. Rollback transaction if local transactional message fails to be committed. * 3. Wait for transaction re-check if unknown exception occurs. * */ transaction.commit(); // transaction.rollback(); // Close the producer when it is no longer needed. producer.close(); } }
For transactional messages, the producer needs to construct a transaction checker to check the intermediate status of abnormal transactions. Three transaction statuses can be returned:
- TransactionResolution.COMMIT: Transaction committed. The consumer can retrieve the message.
- TransactionResolution.ROLLBACK: Transaction rolled back. The message will be discarded and cannot be retrieved.
- TransactionResolution.UNKNOWN: The status cannot be determined and the server is expected to check the message status from the producer again.
Subscribing to Transactional Messages
The code for subscribing to transactional messages is the same as that for subscribing to normal messages.
Feedback
Was this page helpful?
Provide feedbackThank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.
For any further questions, feel free to contact us through the chatbot.
Chatbot