Messaging

Apache RocketMQ Testing: Unit Tests, Embedded Broker, and Testcontainers

HelpMeTest

HelpMeTest

— 5 min read
Apache RocketMQ Testing: Unit Tests, Embedded Broker, and Testcontainers

```tldr Apache RocketMQ testing uses an embedded broker for fast unit tests and a real RocketMQ cluster via Testcontainers for integration tests. Key areas to test: normal producer/consumer flows, ordered messages, delay levels, transaction messages, and consumer group failover. ```

```takeaways **Use rocketmq-test or an embedded broker for unit tests.** The rocketmq-spring-boot-starter includes test utilities for Spring applications.

Ordered messages require testing the queue-level ordering guarantee. Send with MessageQueueSelector and verify consumers receive messages in send order.

Transaction messages have a two-phase commit — test both paths. The broker calls your TransactionListener.checkLocalTransaction() — test the confirm and rollback branches.

Delay levels are fixed (not arbitrary delays). Test with level 1 (1s) in integration tests rather than production levels (minutes) to keep tests fast.

Consumer group failover needs multiple consumer instances. Test that when one consumer dies, its message queue is rebalanced to another consumer in the group. ```

RocketMQ Architecture Recap

RocketMQ separates concerns:

  • NameServer — service discovery (like Kafka's ZooKeeper)
  • Broker — message storage and delivery
  • Producer — publishes messages to topics
  • Consumer — subscribes to topics, pull or push model

For testing, you need both NameServer and Broker running.

Setup: Dependencies

<!-- Spring Boot -->
<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-spring-boot-starter</artifactId>
  <version>2.3.0</version>
</dependency>

<!-- Testing -->
<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-test</artifactId>
  <version>5.2.0</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.testcontainers</groupId>
  <artifactId>testcontainers</artifactId>
  <scope>test</scope>
</dependency>

Embedded RocketMQ (Unit Tests)

import org.apache.rocketmq.test.base.BaseConf;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.namesrv.NamesrvController;

class RocketMQUnitTest {
    private static NamesrvController namesrv;
    private static BrokerController broker;
    private static String nameServerAddr;

    @BeforeAll
    static void startBroker() throws Exception {
        // Start NameServer
        namesrv = NamesrvController.createNamesrvController(new String[]{"-n", "localhost:0"});
        namesrv.start();
        nameServerAddr = "localhost:" + namesrv.getNettyServerConfig().getListenPort();

        // Start Broker
        String brokerConfig = "brokerClusterName=TestCluster\n" +
            "brokerName=broker-a\n" +
            "brokerId=0\n" +
            "namesrvAddr=" + nameServerAddr + "\n" +
            "autoCreateTopicEnable=true\n" +
            "storePathRootDir=/tmp/rocketmq-test\n";

        // Use BrokerController.createBrokerController or test framework utilities
    }

    @AfterAll
    static void stopBroker() throws Exception {
        if (broker != null) broker.shutdown();
        if (namesrv != null) namesrv.shutdown();
    }
}

For simpler setup, use the rocketmq-spring-boot-test integration:

@SpringBootTest
@RocketMQTest
class SpringRocketMQTest {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Test
    void sendAndReceiveMessage() throws Exception {
        String destination = "test-topic:test-tag";
        CountDownLatch latch = new CountDownLatch(1);

        rocketMQTemplate.convertAndSend(destination, "hello rocketmq");

        assertTrue(latch.await(10, TimeUnit.SECONDS));
    }
}

Testcontainers Integration

@Testcontainers
class RocketMQIntegrationTest {
    @Container
    static GenericContainer<?> rocketmq = new GenericContainer<>(
        "apache/rocketmq:5.2.0"
    )
    .withCommand("sh", "-c",
        "mqnamesrv & sleep 5 && mqbroker -n localhost:9876 autoCreateTopicEnable=true"
    )
    .withExposedPorts(9876, 10911)
    .waitingFor(Wait.forLogMessage(".*The broker.*boot success.*", 1)
        .withStartupTimeout(Duration.ofSeconds(90)));

    @Test
    void producerSendsMessage() throws Exception {
        String nameServerAddr = "localhost:" + rocketmq.getMappedPort(9876);

        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
        producer.setNamesrvAddr(nameServerAddr);
        producer.start();

        Message msg = new Message("TestTopic", "TestTag", "hello".getBytes(StandardCharsets.UTF_8));
        SendResult result = producer.send(msg);

        assertEquals(SendStatus.SEND_OK, result.getSendStatus());
        assertNotNull(result.getMsgId());
        producer.shutdown();
    }
}

Ordered Message Testing

RocketMQ guarantees ordering within a single message queue:

@Test
void orderedMessages_receivedInSendOrder() throws Exception {
    DefaultMQProducer producer = new DefaultMQProducer("ordered-group");
    producer.setNamesrvAddr(nameServerAddr);
    producer.start();

    String orderId = "order-42";
    List<String> statuses = List.of("CREATED", "PAID", "SHIPPED", "DELIVERED");

    // Send all messages for the same order to the same queue
    for (String status : statuses) {
        Message msg = new Message("OrderTopic", "status",
            (orderId + ":" + status).getBytes());

        producer.send(msg, (mqs, message, arg) -> {
            // Always route to same queue for same order
            int idx = Math.abs(arg.toString().hashCode()) % mqs.size();
            return mqs.get(idx);
        }, orderId);
    }
    producer.shutdown();

    // Consume and verify order
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ordered-consumer-group");
    consumer.setNamesrvAddr(nameServerAddr);
    consumer.subscribe("OrderTopic", "status");

    List<String> received = Collections.synchronizedList(new ArrayList<>());
    CountDownLatch latch = new CountDownLatch(4);

    consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
        msgs.forEach(m -> {
            received.add(new String(m.getBody()));
            latch.countDown();
        });
        return ConsumeOrderlyStatus.SUCCESS;
    });
    consumer.start();

    assertTrue(latch.await(30, TimeUnit.SECONDS));

    List<String> receivedStatuses = received.stream()
        .map(s -> s.split(":")[1])
        .collect(Collectors.toList());

    assertEquals(statuses, receivedStatuses);
    consumer.shutdown();
}

Delay Level Testing

RocketMQ uses fixed delay levels (not arbitrary durations):

Level 1: 1s, Level 2: 5s, Level 3: 10s, Level 4: 30s ...
Level 18: 2h
@Test
void delayedMessage_deliveredAfterDelay() throws Exception {
    DefaultMQProducer producer = new DefaultMQProducer("delay-group");
    producer.setNamesrvAddr(nameServerAddr);
    producer.start();

    Message msg = new Message("DelayTopic", "delay", "delayed payload".getBytes());
    msg.setDelayTimeLevel(1);  // Level 1 = ~1 second delay

    long sentAt = System.currentTimeMillis();
    producer.send(msg);
    producer.shutdown();

    // Consumer
    AtomicLong receivedAt = new AtomicLong();
    CountDownLatch latch = new CountDownLatch(1);

    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delay-consumer-group");
    consumer.setNamesrvAddr(nameServerAddr);
    consumer.subscribe("DelayTopic", "*");
    consumer.registerMessageListener((MessageListenerConcurrently) (msgs, ctx) -> {
        receivedAt.set(System.currentTimeMillis());
        latch.countDown();
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    consumer.start();

    assertTrue(latch.await(15, TimeUnit.SECONDS));
    long delay = receivedAt.get() - sentAt;
    assertTrue(delay >= 1000, "Should be delayed at least 1s, was: " + delay + "ms");
    consumer.shutdown();
}

Transaction Message Testing

RocketMQ's transactional messages use two-phase commit:

class OrderTransactionListener implements TransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // Execute local DB transaction
            orderService.createOrder((OrderRequest) arg);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // Broker calls this to check transaction status
        String orderId = msg.getUserProperty("orderId");
        return orderService.orderExists(orderId)
            ? LocalTransactionState.COMMIT_MESSAGE
            : LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

// Unit test the listener in isolation
class OrderTransactionListenerTest {
    @Mock OrderService orderService;

    OrderTransactionListener listener;

    @BeforeEach
    void setup() {
        listener = new OrderTransactionListener(orderService);
    }

    @Test
    void executeLocalTransaction_commitsOnSuccess() {
        when(orderService.createOrder(any())).thenReturn(new Order("O-001"));

        Message msg = new Message();
        LocalTransactionState state = listener.executeLocalTransaction(msg, new OrderRequest("O-001"));

        assertEquals(LocalTransactionState.COMMIT_MESSAGE, state);
    }

    @Test
    void executeLocalTransaction_rollsBackOnFailure() {
        doThrow(new RuntimeException("DB error")).when(orderService).createOrder(any());

        LocalTransactionState state = listener.executeLocalTransaction(new Message(), new OrderRequest("fail"));

        assertEquals(LocalTransactionState.ROLLBACK_MESSAGE, state);
    }

    @Test
    void checkLocalTransaction_commitsIfOrderExists() {
        when(orderService.orderExists("O-001")).thenReturn(true);

        MessageExt msg = new MessageExt();
        msg.putUserProperty("orderId", "O-001");
        LocalTransactionState state = listener.checkLocalTransaction(msg);

        assertEquals(LocalTransactionState.COMMIT_MESSAGE, state);
    }
}

Consumer Group Concurrency Testing

@Test
void concurrentConsumers_processAllMessages() throws Exception {
    int messageCount = 50;
    int consumerCount = 3;

    // Start multiple consumers in same group
    List<DefaultMQPushConsumer> consumers = new ArrayList<>();
    Set<String> processedIds = Collections.synchronizedSet(new HashSet<>());
    CountDownLatch latch = new CountDownLatch(messageCount);

    for (int i = 0; i < consumerCount; i++) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("concurrent-group");
        consumer.setNamesrvAddr(nameServerAddr);
        consumer.setConsumeThreadMin(2);
        consumer.setConsumeThreadMax(4);
        consumer.subscribe("ConcurrentTopic", "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, ctx) -> {
            msgs.forEach(m -> {
                processedIds.add(m.getMsgId());
                latch.countDown();
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        consumers.add(consumer);
    }

    // Publish messages
    DefaultMQProducer producer = new DefaultMQProducer("concurrent-producer");
    producer.setNamesrvAddr(nameServerAddr);
    producer.start();
    for (int i = 0; i < messageCount; i++) {
        producer.send(new Message("ConcurrentTopic", "*", ("msg-" + i).getBytes()));
    }
    producer.shutdown();

    assertTrue(latch.await(30, TimeUnit.SECONDS));
    assertEquals(messageCount, processedIds.size(), "Each message should be processed exactly once");

    consumers.forEach(c -> c.shutdown());
}

Common RocketMQ Testing Issues

NameServer not ready — producers/consumers connecting before NameServer finishes starting get connection refused. Add a startup delay or health check.

autoCreateTopicEnable=false in prod but true in tests — if you test with auto-create and deploy without it, topics won't exist in production. Create topics explicitly in tests.

Consumer group reuse across tests — each test should use a unique consumer group name to avoid offset sharing and missed messages.

Store path collisions — embedded broker stores data in a local directory. Use a temp directory per test run: storePathRootDir=/tmp/rocketmq- + UUID.randomUUID().

Read more

Start now free