ApsaraMQ for RocketMQ:トランザクションメッセージ

import java.time.Duration;
import org.apache.rocketmq.client.apis.*;
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.TransactionResolution;
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.shaded.com.google.common.base.Strings;

public class ProducerTransactionMessageExample {

    // データベースに注文が存在するかどうかのチェックをシミュレートします。
    private static boolean checkOrderById(String orderId) {
        return true;
    }

    // ローカルトランザクション (例：注文レコードの挿入) をシミュレートします。
    private static boolean doLocalTransaction() {
        return true;
    }

    public static void main(String[] args) throws ClientException {
        // ご利用のインスタンスエンドポイントに置き換えてください。
        // これは、ApsaraMQ for RocketMQ コンソールの [インスタンス詳細] ページの
        // [エンドポイント] タブで確認できます。
        String endpoints = "<your-instance-endpoint>";

        // トピックの MessageType は Transaction に設定する必要があります。
        String topic = "<your-transaction-topic>";

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints);

        // 認証:
        // - パブリックエンドポイント：ユーザー名とパスワードを指定します (これらは
        //   [アクセス制御] ページの [インテリジェント認証] タで確認できます)。
        // - ECS 上の VPC エンドポイント：認証情報は不要です。ブローカーが VPC から
        //   解決します。
        // - サーバーレスインスタンス：アクセス方法に関係なく、常に認証情報を
        //   指定します。
        builder.setCredentialProvider(
            new StaticSessionCredentialsProvider("<your-username>", "<your-password>"));
        builder.setRequestTimeout(Duration.ofMillis(5000));
        ClientConfiguration configuration = builder.build();

        MessageBuilder messageBuilder = new MessageBuilderImpl();

        // トランザクションチェッカーを使用してプロデューサーをビルドします。
        // このチェッカーは、コミット/ロールバックの結果が受信されなかったハーフメッセージを
        // ブローカーがクエリする際に実行されます。
        Producer producer = provider.newProducerBuilder()
            .setTransactionChecker(messageView -> {
                // ハーフメッセージにアタッチされた注文 ID を検索します。
                // 注文がデータベースに存在する場合、ローカルトランザクションは
                // 正常にコミットされています。それ以外の場合は、ロールバックします。
                final String orderId = messageView.getProperties().get("OrderId");
                if (Strings.isNullOrEmpty(orderId)) {
                    return TransactionResolution.ROLLBACK;
                }
                return checkOrderById(orderId)
                    ? TransactionResolution.COMMIT
                    : TransactionResolution.ROLLBACK;
            })
            .setTopics(topic)
            .setClientConfiguration(configuration)
            .build();

        // ステップ 1：トランザクションを開始します。
        final Transaction transaction;
        try {
            transaction = producer.beginTransaction();
        } catch (ClientException e) {
            e.printStackTrace();
            return;
        }

        // ステップ 2：ハーフメッセージをビルドして送信します。
        Message message = messageBuilder.setTopic(topic)
            .setKeys("messageKey1")
            .setTag("messageTag")
            // 後でトランザクションチェッカーがクエリできるように、ビジネス ID をアタッチします。
            .addProperty("OrderId", "xxx")
            .setBody("messageBody".getBytes())
            .build();

        final SendReceipt sendReceipt;
        try {
            sendReceipt = producer.send(message, transaction);
        } catch (ClientException e) {
            // ハーフメッセージの送信に失敗しました。トランザクションはここで終了します。
            return;
        }

        // ステップ 3：ローカルトランザクションを実行します。
        boolean localTransactionOk = doLocalTransaction();

        // ステップ 4：ローカルトランザクションの結果に基づいてコミットまたはロールバックします。
        if (localTransactionOk) {
            try {
                transaction.commit();
            } catch (ClientException e) {
                // コミット呼び出しが失敗した場合、ブローカーは
                // トランザクションチェッカーを呼び出してステータスを解決します。
                e.printStackTrace();
            }
        } else {
            try {
                transaction.rollback();
            } catch (ClientException e) {
                // エラーをログに記録します。ブローカーはトランザクションチェッカーを
                // 呼び出してステータスを解決します。
                e.printStackTrace();
            }
        }
    }
}