tencent cloud

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

public class ProducerTransactionMessageDemo {

    private static final Logger log = LoggerFactory.getLogger(ProducerTransactionMessageDemo.class);

    private static boolean executeLocalTransaction() {
        // Simulate a local transaction (such as a database insert operation), assume execution successful here
        return true;
    }

    private static boolean checkTransactionStatus(String orderId) {
        // Simulate local query of transaction execution result, for example, query whether the order ID has been stored, return true if found
        return true;
    }


    public static void main(String[] args) throws ClientException {

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        // Get ak and sk in the console permission management page
        String accessKey = "your-ak";
        String secretKey = "your-sk";
        SessionCredentialsProvider sessionCredentialsProvider =
            new StaticSessionCredentialsProvider(accessKey, secretKey);

        // Get the access address provided by Tencent Cloud from console and fill in
        String endpoints = "https://your-endpoints";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .enableSsl(false)
            .setCredentialProvider(sessionCredentialsProvider)
            .build();

        String topic = "tran_topic";
        TransactionChecker checker = messageView -> {
            log.info("Receive transactional result check request, message={}", messageView);
            // Server proactively checks local transaction status
            String orderId = messageView.getProperties().get("orderId");
            boolean isSuccess = checkTransactionStatus(orderId);
            return isSuccess ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;
        };

        Create a producer and set the checker object
        Producer producer = provider.newProducerBuilder()
            .setClientConfiguration(clientConfiguration)
            .setTopics(topic)
            .setTransactionChecker(checker)
            .build();

        execute a transaction enabling statement
        final Transaction transaction = producer.beginTransaction();

        byte[] body = "This is a transaction message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
        String tag = "tagA";
        final Message message = provider.newMessageBuilder()
            .setTopic(topic)
            .setTag(tag)
            .setKeys("your-key-565ef26f5727")
            //Transactional messages usually set a unique ID associated with the local transaction, used to validate local transaction operations.
            .addProperty("orderId", "0001")
            .setBody(body)
            .build();

        // Send half message
        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 local transaction
        boolean localTxSuccess = executeLocalTransaction();
        if (localTxSuccess) {
            // Local transaction is successfully executed, secondary confirmation is Commit
            transaction.commit();
        } else {
            // Local transaction execution failed, secondary confirmation is Rollback
            transaction.rollback();
        }
        // producer.close();
    }
}


private static boolean executeLocalTransaction() {
    // Local transaction execution failed
    return false;
}

private static boolean checkTransactionStatus(String orderId) {
    // The result is also rollback, return false
    return false;
}