Testing

Testing Event-Driven Microservices: Kafka, RabbitMQ, and Beyond

HelpMeTest

HelpMeTest

— 8 min read
Testing Event-Driven Microservices: Kafka, RabbitMQ, and Beyond

Testing asynchronous, event-driven systems is genuinely hard. The problems stack up fast: you can't assert response.status_code == 200 when the operation happens asynchronously; your test needs to know when to stop waiting; race conditions create flaky tests that pass locally and fail in CI; and the message broker is a shared piece of infrastructure that makes tests brittle and slow to run in isolation.

Yet event-driven architectures built on Kafka, RabbitMQ, or similar brokers are increasingly common — and leaving them undertested is a reliability risk you'll pay for in production incidents. This guide covers practical testing strategies for event-driven systems, from unit testing individual producers and consumers to integration testing with embedded brokers using Testcontainers.

The Core Challenges of Async Testing

Before diving into solutions, it helps to name the specific challenges:

The timing problem — After triggering an event, when do you check for the result? Sleep-based waiting (time.sleep(2)) is fragile and slow. Too short and tests fail intermittently; too long and your test suite takes forever.

The isolation problem — If your Kafka broker is shared across tests, messages from one test can bleed into another. Consumer group offsets, retained messages, and partition state all create hidden coupling between test cases.

The observation problem — In a synchronous system, you call a function and check its return value. In an async system, the "return value" is a side effect — a message in a different topic, a database row written, an HTTP call made. You need to know where to look.

The infrastructure problem — Tests that depend on a running Kafka cluster are slow to start, expensive to provision in CI, and painful to reset between test runs.

Each of these has a good solution. Let's work through them.

Unit Testing Producers and Consumers

The first layer of testing doesn't need a broker at all. Producer and consumer logic can be tested in isolation by mocking the broker client.

Here's a Kafka producer for an order service:

// order-event-producer.js
class OrderEventProducer {
  constructor(kafkaClient) {
    this.producer = kafkaClient.producer({
      idempotent: true,
      maxInFlightRequests: 1,
    });
  }

  async connect() {
    await this.producer.connect();
  }

  async publishOrderCreated(order) {
    const event = {
      eventType: 'order.created',
      eventId: crypto.randomUUID(),
      timestamp: new Date().toISOString(),
      version: '1.0',
      data: {
        orderId: order.id,
        customerId: order.customerId,
        totalAmount: order.totalAmount,
        items: order.items,
        status: 'pending',
      },
    };

    await this.producer.send({
      topic: 'order-events',
      messages: [
        {
          key: order.id,                        // partition by order ID for ordering
          value: JSON.stringify(event),
          headers: {
            'content-type': 'application/json',
            'event-type': 'order.created',
          },
        },
      ],
    });

    return event;
  }

  async publishOrderCancelled(orderId, reason) {
    const event = {
      eventType: 'order.cancelled',
      eventId: crypto.randomUUID(),
      timestamp: new Date().toISOString(),
      data: { orderId, reason },
    };

    await this.producer.send({
      topic: 'order-events',
      messages: [{ key: orderId, value: JSON.stringify(event) }],
    });
  }
}

Unit test it without any broker:

// order-event-producer.test.js
const { OrderEventProducer } = require('./order-event-producer');

describe('OrderEventProducer', () => {
  let mockProducer;
  let mockKafkaClient;
  let producer;

  beforeEach(() => {
    mockProducer = {
      connect: jest.fn().mockResolvedValue(undefined),
      send: jest.fn().mockResolvedValue([{ topicName: 'order-events', partition: 0, errorCode: 0 }]),
    };
    mockKafkaClient = {
      producer: jest.fn().mockReturnValue(mockProducer),
    };
    producer = new OrderEventProducer(mockKafkaClient);
  });

  it('publishes order.created event with correct structure', async () => {
    await producer.connect();
    const order = {
      id: 'ord_123',
      customerId: 'cust_456',
      totalAmount: 99.99,
      items: [{ sku: 'PROD-1', quantity: 2, price: 49.99 }],
    };

    const event = await producer.publishOrderCreated(order);

    expect(mockProducer.send).toHaveBeenCalledTimes(1);
    const sendCall = mockProducer.send.mock.calls[0][0];
    
    expect(sendCall.topic).toBe('order-events');
    expect(sendCall.messages).toHaveLength(1);
    
    const message = sendCall.messages[0];
    expect(message.key).toBe('ord_123');  // keyed by order ID for partition ordering
    
    const payload = JSON.parse(message.value);
    expect(payload.eventType).toBe('order.created');
    expect(payload.version).toBe('1.0');
    expect(payload.data.orderId).toBe('ord_123');
    expect(payload.data.totalAmount).toBe(99.99);
    expect(payload.eventId).toMatch(/^[0-9a-f-]{36}$/);  // UUID format
    expect(new Date(payload.timestamp)).toBeInstanceOf(Date);  // valid ISO timestamp
  });

  it('uses idempotent producer configuration', () => {
    expect(mockKafkaClient.producer).toHaveBeenCalledWith(
      expect.objectContaining({ idempotent: true })
    );
  });
});

Now the consumer side. Here's an inventory service consumer that reacts to order events:

# inventory_consumer.py
import json
import logging
from kafka import KafkaConsumer
from inventory_repository import InventoryRepository

class InventoryConsumer:
    def __init__(self, consumer: KafkaConsumer, repository: InventoryRepository):
        self.consumer = consumer
        self.repository = repository
        self.logger = logging.getLogger(__name__)

    def process_message(self, message):
        """Process a single Kafka message. Extracted for testability."""
        try:
            event = json.loads(message.value)
        except json.JSONDecodeError as e:
            self.logger.error(f"Failed to parse message: {e}")
            return {"status": "parse_error", "error": str(e)}

        event_type = event.get("eventType")
        
        if event_type == "order.created":
            return self._handle_order_created(event)
        elif event_type == "order.cancelled":
            return self._handle_order_cancelled(event)
        else:
            self.logger.warning(f"Unknown event type: {event_type}")
            return {"status": "ignored", "reason": f"unknown event type: {event_type}"}

    def _handle_order_created(self, event):
        order_id = event["data"]["orderId"]
        items = event["data"]["items"]
        
        for item in items:
            reserved = self.repository.reserve(item["sku"], item["quantity"])
            if not reserved:
                self.logger.error(
                    f"Insufficient inventory for {item['sku']}, order {order_id}"
                )
                return {
                    "status": "insufficient_inventory",
                    "sku": item["sku"],
                    "order_id": order_id,
                }
        
        return {"status": "reserved", "order_id": order_id}

    def _handle_order_cancelled(self, event):
        order_id = event["data"]["orderId"]
        self.repository.release_reservation(order_id)
        return {"status": "released", "order_id": order_id}

Unit test the consumer logic without a broker:

# test_inventory_consumer.py
import json
from unittest.mock import MagicMock, create_autospec
from inventory_consumer import InventoryConsumer
from inventory_repository import InventoryRepository

def make_message(event_dict):
    """Create a mock Kafka message."""
    msg = MagicMock()
    msg.value = json.dumps(event_dict).encode()
    msg.offset = 100
    msg.partition = 0
    return msg

class TestInventoryConsumer:
    def setup_method(self):
        self.repo = create_autospec(InventoryRepository)
        self.consumer = InventoryConsumer(consumer=MagicMock(), repository=self.repo)

    def test_reserves_inventory_on_order_created(self):
        self.repo.reserve.return_value = True
        
        message = make_message({
            "eventType": "order.created",
            "data": {
                "orderId": "ord_123",
                "items": [
                    {"sku": "PROD-1", "quantity": 2},
                    {"sku": "PROD-2", "quantity": 1},
                ],
            },
        })
        
        result = self.consumer.process_message(message)
        
        assert result["status"] == "reserved"
        assert self.repo.reserve.call_count == 2
        self.repo.reserve.assert_any_call("PROD-1", 2)
        self.repo.reserve.assert_any_call("PROD-2", 1)

    def test_returns_error_when_inventory_insufficient(self):
        self.repo.reserve.return_value = False  # Simulates out-of-stock
        
        message = make_message({
            "eventType": "order.created",
            "data": {
                "orderId": "ord_456",
                "items": [{"sku": "SOLD-OUT-SKU", "quantity": 1}],
            },
        })
        
        result = self.consumer.process_message(message)
        
        assert result["status"] == "insufficient_inventory"
        assert result["sku"] == "SOLD-OUT-SKU"

    def test_ignores_unknown_event_types(self):
        message = make_message({"eventType": "product.updated", "data": {}})
        result = self.consumer.process_message(message)
        assert result["status"] == "ignored"

    def test_handles_malformed_json_gracefully(self):
        msg = MagicMock()
        msg.value = b"not valid json {"
        
        result = self.consumer.process_message(msg)
        
        assert result["status"] == "parse_error"
        self.repo.reserve.assert_not_called()

Integration Testing with Testcontainers

Unit tests verify logic; integration tests verify that your code works with a real broker. Testcontainers spins up a real Kafka (or RabbitMQ) in Docker for each test run, giving you genuine isolation without a shared broker.

// OrderEventIntegrationTest.java
@SpringBootTest
@Testcontainers
class OrderEventIntegrationTest {

    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.4.0")
    );

    @DynamicPropertySource
    static void kafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }

    @Autowired
    private OrderService orderService;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private InventoryRepository inventoryRepository;

    @Test
    void orderCreation_publishesEventAndUpdatesInventory() throws Exception {
        // Set up initial inventory
        inventoryRepository.setStock("PROD-1", 10);
        
        // Create an order (triggers event production internally)
        Order order = orderService.createOrder(
            "cust_123",
            List.of(new OrderItem("PROD-1", 2, new BigDecimal("29.99")))
        );
        
        // Wait for the event to be consumed and inventory updated
        await()
            .atMost(Duration.ofSeconds(15))
            .pollInterval(Duration.ofMillis(500))
            .untilAsserted(() -> {
                int remainingStock = inventoryRepository.getStock("PROD-1");
                assertThat(remainingStock).isEqualTo(8);  // 10 - 2
            });
        
        // Verify the event was published with correct structure
        KafkaConsumer<String, String> verifyConsumer = createTestConsumer("verify-group");
        verifyConsumer.subscribe(List.of("order-events"));
        
        ConsumerRecords<String, String> records = verifyConsumer.poll(Duration.ofSeconds(5));
        assertThat(records.count()).isGreaterThanOrEqualTo(1);
        
        ConsumerRecord<String, String> record = records.iterator().next();
        assertThat(record.key()).isEqualTo(order.getId());
        
        JsonNode event = objectMapper.readTree(record.value());
        assertThat(event.get("eventType").asText()).isEqualTo("order.created");
        assertThat(event.get("data").get("orderId").asText()).isEqualTo(order.getId());
        
        verifyConsumer.close();
    }

    private KafkaConsumer<String, String> createTestConsumer(String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId + "-" + UUID.randomUUID());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new KafkaConsumer<>(props);
    }
}

The await() call uses the Awaitility library for condition-based waiting — far better than sleeping. It polls every 500ms and fails with a clear message after 15 seconds, rather than hanging indefinitely or failing too early.

Testing Consumer Group Behavior

Consumer groups are easy to misconfigure — wrong group IDs cause every instance to read all partitions, leading to duplicate processing. Test this explicitly:

@Test
void multipleConsumerInstances_processEachMessageOnlyOnce() throws Exception {
    String topic = "inventory-events";
    int messageCount = 20;
    
    // Send 20 messages
    for (int i = 0; i < messageCount; i++) {
        kafkaTemplate.send(topic, "key-" + i, 
            buildInventoryEvent("PROD-" + i, 5)
        );
    }
    
    // Count how many times each message was processed
    AtomicInteger processCount = new AtomicInteger(0);
    Set<String> processedKeys = ConcurrentHashMap.newKeySet();
    Set<String> duplicates = ConcurrentHashMap.newKeySet();
    
    // Register an observer on both consumer instances
    inventoryConsumer1.onMessageProcessed(record -> {
        if (!processedKeys.add(record.key())) {
            duplicates.add(record.key());
        }
        processCount.incrementAndGet();
    });
    inventoryConsumer2.onMessageProcessed(record -> {
        if (!processedKeys.add(record.key())) {
            duplicates.add(record.key());
        }
        processCount.incrementAndGet();
    });
    
    // Wait for all messages to be processed
    await()
        .atMost(Duration.ofSeconds(30))
        .until(() -> processCount.get() >= messageCount);
    
    // Allow a small window for any potential duplicates
    Thread.sleep(2000);
    
    assertThat(processCount.get()).isEqualTo(messageCount);
    assertThat(duplicates).isEmpty();
}

Testing Dead Letter Queues

Dead letter queues (DLQs) are your safety net for unprocessable messages. If they're not tested, you won't know they work until a production incident. Here's how to test DLQ routing for both Kafka and RabbitMQ:

# test_dead_letter_queue.py

class TestDeadLetterQueue:
    def test_poison_messages_route_to_dlq(self, kafka_container):
        """Messages that fail processing 3 times should appear on the DLQ topic."""
        admin = AdminClient({"bootstrap.servers": kafka_container.bootstrap_servers})
        
        # Send a message that will always fail processing (invalid schema)
        poison_message = b'{"eventType": "order.created", "data": null}'  # null data causes NPE
        
        producer = Producer({"bootstrap.servers": kafka_container.bootstrap_servers})
        producer.produce("order-events", value=poison_message, key=b"bad-order")
        producer.flush()
        
        # The consumer should retry 3 times then send to DLQ
        dlq_consumer = Consumer({
            "bootstrap.servers": kafka_container.bootstrap_servers,
            "group.id": "dlq-verifier",
            "auto.offset.reset": "earliest",
        })
        dlq_consumer.subscribe(["order-events.DLQ"])
        
        # Wait for DLQ message (allow time for 3 retries + routing)
        deadline = time.time() + 30
        dlq_message = None
        while time.time() < deadline:
            msg = dlq_consumer.poll(timeout=1.0)
            if msg and not msg.error():
                dlq_message = msg
                break
        
        assert dlq_message is not None, "Poison message did not appear on DLQ within 30s"
        assert dlq_message.key() == b"bad-order"
        
        # DLQ message should include failure metadata in headers
        headers = dict(dlq_message.headers() or [])
        assert b"x-exception-message" in headers or b"x-death" in headers
        
        dlq_consumer.close()

    def test_valid_messages_never_reach_dlq(self, kafka_container):
        """Valid messages should never appear on the DLQ."""
        valid_message = json.dumps({
            "eventType": "order.created",
            "eventId": str(uuid.uuid4()),
            "timestamp": datetime.utcnow().isoformat(),
            "data": {
                "orderId": "ord_789",
                "customerId": "cust_123",
                "items": [{"sku": "PROD-1", "quantity": 1}],
            },
        }).encode()
        
        producer = Producer({"bootstrap.servers": kafka_container.bootstrap_servers})
        producer.produce("order-events", value=valid_message, key=b"ord_789")
        producer.flush()
        
        # Wait for processing to complete
        time.sleep(5)
        
        # Check DLQ — should be empty
        dlq_consumer = Consumer({
            "bootstrap.servers": kafka_container.bootstrap_servers,
            "group.id": "dlq-check-" + str(uuid.uuid4()),
            "auto.offset.reset": "earliest",
        })
        dlq_consumer.subscribe(["order-events.DLQ"])
        
        msg = dlq_consumer.poll(timeout=3.0)
        assert msg is None or msg.error(), "Valid message incorrectly routed to DLQ"
        
        dlq_consumer.close()

RabbitMQ Testing Patterns

For RabbitMQ-based systems, Testcontainers provides a ready-to-use container:

@Container
static RabbitMQContainer rabbitmq = new RabbitMQContainer(
    DockerImageName.parse("rabbitmq:3.12-management")
);

@Test
void messagePublishedToExchangeRoutesToCorrectQueues() throws Exception {
    // Set up channel and declare exchange/queue topology
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri(rabbitmq.getAmqpUrl());
    
    try (Connection connection = factory.newConnection();
         Channel channel = connection.createChannel()) {
        
        channel.exchangeDeclare("order-exchange", "topic", true);
        channel.queueDeclare("inventory-queue", true, false, false, null);
        channel.queueDeclare("notification-queue", true, false, false, null);
        channel.queueBind("inventory-queue", "order-exchange", "order.created");
        channel.queueBind("notification-queue", "order-exchange", "order.*");
        
        // Publish an order.created event
        String event = objectMapper.writeValueAsString(Map.of(
            "eventType", "order.created",
            "orderId", "ord_123"
        ));
        channel.basicPublish("order-exchange", "order.created", null, event.getBytes());
        
        // Both queues should receive the message
        Thread.sleep(1000);
        
        GetResponse inventoryMsg = channel.basicGet("inventory-queue", false);
        GetResponse notificationMsg = channel.basicGet("notification-queue", false);
        
        assertThat(inventoryMsg).isNotNull();
        assertThat(notificationMsg).isNotNull();
        
        String inventoryEvent = new String(inventoryMsg.getBody());
        assertThat(objectMapper.readTree(inventoryEvent).get("orderId").asText())
            .isEqualTo("ord_123");
    }
}

Strategies for Reliable Async Assertions

A few patterns that eliminate flaky async tests:

Use Awaitility or equivalent — Never use sleep() to wait for async operations. Poll with a timeout and clear failure messages:

// Good
await().atMost(10, SECONDS).until(() -> inventoryRepository.getStock("PROD-1") == 8);

// Bad — arbitrary sleep, either too long or too short
Thread.sleep(3000);
assertEquals(8, inventoryRepository.getStock("PROD-1"));

Unique keys per test — Use UUID-based keys/IDs to prevent message bleed between tests:

order_id = f"test-{uuid.uuid4()}"  # Never reuse IDs across tests

Consumer group isolation — Give each test run a unique consumer group ID so tests don't share offsets and inadvertently skip messages.

Dedicated verification topics — For complex systems, have your services publish to a test-observable topic (only in test mode) that contains structured verification events. Your tests consume from this topic to verify outcomes without needing to query databases directly.

Event-driven architectures reward the investment in async testing infrastructure. Once you have Testcontainers-based Kafka integration tests running reliably in CI, the confidence they provide — especially around DLQ routing, consumer group behavior, and event schema correctness — is hard to get any other way.

Read more