<!-- in your <dependencies> block -->
  <dependency>
      <groupId>org.apache.rocketmq</groupId>
      <artifactId>rocketmq-client</artifactId>
      <version>4.9.4</version>
  </dependency>
  
  <dependency>
      <groupId>org.apache.rocketmq</groupId>
      <artifactId>rocketmq-acl</artifactId>
      <version>4.9.4</version>
  </dependency>

public class TransactionListenerImpl implements TransactionListener {

    //After half messages are sent successfully, call back this method to execute the local transaction.
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
      //Execute the database transaction here. If it is successful, success is returned; otherwise, unknown is returned, or a rollback is performed, waiting for recheck.
      return LocalTransactionState.UNKNOW;
    }
    //Recheck the local transaction.
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
       //Query the data status of the local database here and then determine whether to submit the local database data status.
       return LocalTransactionState.COMMIT_MESSAGE;
    }

}

//The user is required to implement one TransactionListener instance.
TransactionListener transactionListener = new TransactionListenerImpl();
// Instantiate the transactional message producer.
ProducerTransactionMQProducer producer = new TransactionMQProducer("transaction_group",        
// Access control list (ACL) permissions.        
new AclClientRPCHook(new SessionCredentials(ClientCreater.ACCESS_KEY, ClientCreater.SECRET_KEY)));
// Set the NameServer address.
producer.setNamesrvAddr(ClientCreater.NAMESERVER);
producer.setTransactionListener(transactionListener);
producer.start();

for (int i = 0; i < 3; i++) {    
    // Message construction example.    
    Message msg = new Message(TOPIC_NAME, "your tag", "KEY" + i,("Hello RocketMQ " + i).getBytes(StandardCharsets.UTF_8));    
    SendResult sendResult = producer.sendMessageInTransaction(msg,null);    
     System.out.printf("%s%n", sendResult);
}

// Instantiate a consumer.
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(
            groupName,                                              
            new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); // ACL permissions.
        // Set the NameServer address.
        pushConsumer.setNamesrvAddr(nameserver);
        pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { 
           // Message processing logic.    
           System.out.printf("%s Receive transaction messages: %s %n", Thread.currentThread().getName(), msgs);    
           // Mark the message as successfully consumed.    
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
           });

// Subscribe to a topic.
      pushConsumer.subscribe(topic_name, "*");
      // Register a callback implementation class to handle messages pulled from the broker.
      pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
          // Message processing logic.
          System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
          // Mark the message as successfully consumed and return an appropriate status.
          return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      });
      // Start the consumer instance.
      pushConsumer.start();