Event Consumer

A consumer of messages from Kafka, SQS, RabbitMQ, Pub/Sub, or similar. Reads messages, processes them, often updates state and produces downstream messages. The “public interface” is the topic or queue and the schema of messages on it.

This pattern has problems the API provider and API consumer patterns don’t have: ordering, replay, poison messages, dead-letter queues, and delivery semantics (at-most-once, at-least-once, exactly-once-with-effort).

What needs covered

Positive test cases

Common cases to consider, not an exhaustive list. Drop items that don’t apply and add ones the pattern doesn’t mention but your component needs.

• Well-formed message: produces the expected state change and the documented downstream events.
• Batch processing: processes per documented policy.
• Replay from offset: reproduces the same end state.
• Documented schema versions: are accepted.

Negative test cases

Common cases to consider, not an exhaustive list. Drop items that don’t apply and add ones the pattern doesn’t mention but your component needs.

• Malformed message: routes to the DLQ with a correlation ID; the consumer survives.
• Duplicate delivery: absorbed by idempotency.
• Out-of-order delivery: follows the documented behavior.
• Mid-batch downstream failure: the offset is left uncommitted.
• Schema-version skew: handled per the documented policy.
• Slow downstream: applies backpressure rather than OOM.
• Consumer-group rebalance during processing: no in-flight messages are stranded.

Test double validation

Adapter integration tests validate the broker double in component tests. They run against a real broker container the team controls (Kafka in Docker, ElasticMQ for SQS, Redpanda in Docker). The test asserts the broker client adapter speaks the protocol correctly against that controlled instance. It does not assert which messages the broker returns or in what order, because that is the broker’s behavior, not the adapter’s. Contract tests pinning each version validate the schema registry double, plus a post-deploy check against the real registry. Post-deploy synthetic publishes a known message to the real topic in a non-prod environment.

Pipeline placement

Handler unit tests and component tests run in CI Stage 1. Adapter integration tests against a team-controlled broker container run in CI Stage 1 or Stage 2. Some adapter integration tests run against a managed broker the team can’t pin to a known state. Those tests run out-of-band on a schedule, alongside the post-deploy synthetic.

Example: idempotency under duplicate delivery

Money.usd takes minor units (cents); 4250 represents $42.50.

• Java
• C#
• JavaScript

@Test
void same_message_processed_twice_creates_one_payment_record() {
  PaymentEvent event = new PaymentEvent(
      "evt-9f12", OrderId.of("ord-001"), Money.usd(4250));
  PaymentRepo repo = new InMemoryPaymentRepo();
  PaymentEventHandler handler = new PaymentEventHandler(repo);

  handler.handle(event);
  handler.handle(event);

  assertThat(repo.findByEventId("evt-9f12")).hasSize(1);
  assertThat(repo.totalForOrder(OrderId.of("ord-001"))).isEqualTo(Money.usd(4250));
}

[Fact]
public void Same_message_processed_twice_creates_one_payment_record()
{
    var evt = new PaymentEvent("evt-9f12", OrderId.Of("ord-001"), Money.Usd(4250));
    var repo = new InMemoryPaymentRepo();
    var handler = new PaymentEventHandler(repo);

    handler.Handle(evt);
    handler.Handle(evt);

    repo.FindByEventId("evt-9f12").Should().HaveCount(1);
    repo.TotalForOrder(OrderId.Of("ord-001")).Should().Be(Money.Usd(4250));
}

test("same message processed twice creates one payment record", () => {
  const event = new PaymentEvent(
    "evt-9f12", OrderId.of("ord-001"), Money.usd(4250));
  const repo = new InMemoryPaymentRepo();
  const handler = new PaymentEventHandler(repo);

  handler.handle(event);
  handler.handle(event);

  expect(repo.findByEventId("evt-9f12")).toHaveLength(1);
  expect(repo.totalForOrder(OrderId.of("ord-001"))).toEqual(Money.usd(4250));
});