Testing

Contract Testing Async Messaging with Pact: Kafka and SQS

HelpMeTest

HelpMeTest

— 5 min read
Contract Testing Async Messaging with Pact: Kafka and SQS

HTTP contract testing is well understood. You define the request and response, Pact records them as a contract, and the provider verifies it. But microservices don't only communicate over HTTP — they publish and consume events through Kafka topics, SQS queues, and SNS topics. Those async channels need contracts too.

Pact supports message contracts with a different model: instead of HTTP request/response pairs, you define message format contracts. The consumer declares what structure it expects to receive. The provider (the publisher) verifies it produces messages that match.

Message Contracts vs HTTP Contracts

HTTP contracts capture a full request/response cycle. Message contracts capture only the message body and metadata — there's no "request" because the consumer didn't send one.

The consumer test answers: "given a message arrives on this topic, can my handler process it correctly?"

The provider test answers: "does our publisher produce messages with the structure consumers expect?"

The pact file for a message contract looks like:

{
  "consumer": {"name": "inventory-service"},
  "provider": {"name": "order-service"},
  "messages": [
    {
      "description": "an order placed event",
      "providerStates": [{"name": "an order exists for customer 123"}],
      "contents": {
        "orderId": "ord-42",
        "customerId": "cust-123",
        "items": [
          {"productId": "prod-7", "quantity": 2, "unitPrice": 29.99}
        ],
        "totalAmount": 59.98,
        "placedAt": "2024-01-15T10:30:00Z"
      },
      "matchingRules": {
        "body": {
          "$.orderId": {"matchers": [{"match": "type"}]},
          "$.customerId": {"matchers": [{"match": "type"}]},
          "$.items": {"matchers": [{"match": "type", "min": 1}]},
          "$.totalAmount": {"matchers": [{"match": "decimal"}]},
          "$.placedAt": {"matchers": [{"match": "timestamp", "format": "yyyy-MM-dd'T'HH:mm:ssZ"}]}
        }
      },
      "metadata": {
        "contentType": "application/json"
      }
    }
  ]
}

Consumer Message Tests in Node.js

The consumer (inventory-service) handles OrderPlaced events and updates its stock records.

// src/__tests__/order-events-consumer.test.js
const { MessageConsumerPact, asynchronousBodyHandler } = require('@pact-foundation/pact');
const { like, eachLike, decimal, timestamp } = require('@pact-foundation/pact').Matchers;
const path = require('path');
const { handleOrderPlaced } = require('../handlers/order-placed-handler');

describe('OrderPlaced event consumer', () => {
  const messagePact = new MessageConsumerPact({
    consumer: 'inventory-service',
    provider: 'order-service',
    dir: path.resolve(process.cwd(), 'pacts'),
    logLevel: 'warn',
  });

  describe('handles an OrderPlaced event', () => {
    it('updates inventory for each item in the order', () => {
      return messagePact
        .given('an order exists for customer 123')
        .expectsToReceive('an order placed event')
        .withContent({
          orderId: like('ord-42'),
          customerId: like('cust-123'),
          items: eachLike({
            productId: like('prod-7'),
            quantity: like(2),
            unitPrice: decimal(29.99),
          }),
          totalAmount: decimal(59.98),
          placedAt: timestamp("yyyy-MM-dd'T'HH:mm:ssZ", '2024-01-15T10:30:00Z'),
        })
        .withMetadata({ contentType: 'application/json' })
        .verify(asynchronousBodyHandler(handleOrderPlaced));
    });
  });
});

The asynchronousBodyHandler wrapper calls your handler with the message body and verifies it doesn't throw. Your handler:

// src/handlers/order-placed-handler.js
async function handleOrderPlaced(message) {
  const { orderId, items } = message;

  for (const item of items) {
    await decrementStock(item.productId, item.quantity);
  }

  await markOrderFulfillable(orderId);
}

module.exports = { handleOrderPlaced };

This is the key insight of message consumer tests: you test your handler with the message shape defined in the contract. No Kafka broker needed. No SQS mock. Just your handler and a JSON object.

Provider Message Tests: Verifying Kafka Publishers

On the provider side (order-service), you need to prove that your Kafka publisher produces messages matching the contract.

// src/__tests__/order-events-provider.test.js
const { MessageProviderPact } = require('@pact-foundation/pact');
const path = require('path');
const { buildOrderPlacedMessage } = require('../events/order-placed-builder');
const { seedOrder } = require('../test-helpers/seed');

describe('Order service message provider verification', () => {
  const provider = new MessageProviderPact({
    provider: 'order-service',
    pactBrokerUrl: process.env.PACT_BROKER_BASE_URL,
    pactBrokerToken: process.env.PACT_BROKER_TOKEN,
    publishVerificationResults: true,
    providerVersion: process.env.GITHUB_SHA || 'local',
    providerVersionBranch: process.env.GITHUB_REF_NAME || 'main',
    // Map "provider state" names to setup functions
    stateHandlers: {
      'an order exists for customer 123': async () => {
        await seedOrder({
          id: 'ord-42',
          customerId: 'cust-123',
          items: [{ productId: 'prod-7', quantity: 2, unitPrice: 29.99 }],
        });
      },
    },
    // Map "description" strings to message builder functions
    messageProviders: {
      'an order placed event': async () => {
        const order = await getOrder('ord-42');
        return buildOrderPlacedMessage(order);
      },
    },
  });

  it('satisfies all message consumer contracts', () => {
    return provider.verify();
  });
});

The messageProviders map is what makes this work. For each message description in the contract, you return the actual payload your service would publish. Pact compares it against the consumer's expectations using the matchers in the pact file.

// src/events/order-placed-builder.js
function buildOrderPlacedMessage(order) {
  return {
    orderId: order.id,
    customerId: order.customerId,
    items: order.items.map(item => ({
      productId: item.productId,
      quantity: item.quantity,
      unitPrice: item.unitPrice,
    })),
    totalAmount: order.items.reduce((sum, item) => sum + item.quantity * item.unitPrice, 0),
    placedAt: order.createdAt.toISOString(),
  };
}

module.exports = { buildOrderPlacedMessage };

Testing Kafka Consumers with Pact

The consumer test above doesn't involve Kafka at all — it tests the handler in isolation. But you also want to test that your Kafka consumer wiring correctly deserializes messages and routes them to the handler.

Separate that concern into an integration test:

// src/__tests__/kafka-consumer-integration.test.js
const { Kafka } = require('kafkajs');
const { handleOrderPlaced } = require('../handlers/order-placed-handler');

// Use testcontainers or a local Kafka for this test
describe('Kafka consumer wiring', () => {
  let kafka;
  let producer;
  let consumer;

  beforeAll(async () => {
    kafka = new Kafka({
      clientId: 'test-client',
      brokers: [process.env.KAFKA_BROKER || 'localhost:9092'],
    });
    producer = kafka.producer();
    consumer = kafka.consumer({ groupId: 'inventory-test-group' });
    await producer.connect();
    await consumer.connect();
    await consumer.subscribe({ topic: 'order.placed', fromBeginning: false });
  });

  afterAll(async () => {
    await producer.disconnect();
    await consumer.disconnect();
  });

  it('deserializes and routes OrderPlaced events', async () => {
    const received = [];

    await consumer.run({
      eachMessage: async ({ message }) => {
        const payload = JSON.parse(message.value.toString());
        await handleOrderPlaced(payload);
        received.push(payload);
      },
    });

    await producer.send({
      topic: 'order.placed',
      messages: [{
        key: 'ord-100',
        value: JSON.stringify({
          orderId: 'ord-100',
          customerId: 'cust-456',
          items: [{ productId: 'prod-3', quantity: 1, unitPrice: 15.00 }],
          totalAmount: 15.00,
          placedAt: new Date().toISOString(),
        }),
      }],
    });

    // Wait for the message to be processed
    await new Promise(resolve => setTimeout(resolve, 1000));
    expect(received).toHaveLength(1);
    expect(received[0].orderId).toBe('ord-100');
  });
});

The contract test and the Kafka integration test serve different purposes:

  • Contract test: proves the message shape is compatible with what consumers expect
  • Kafka integration test: proves the wiring (deserialization, routing, error handling) works

SQS Contract Testing Patterns

SQS follows the same pattern. The contract is about the message body, not the transport.

Consumer test (Python handler):

# tests/test_order_events_consumer.py
import pytest
import json
from pact import MessageConsumerPact, Format
from src.handlers.order_handler import handle_order_placed

pact = MessageConsumerPact(
    consumer="fulfillment-service",
    provider="order-service",
    pact_dir="./pacts",
    log_level="WARNING",
)

def test_handles_order_placed_event():
    (
        pact
        .given("an order exists")
        .expects_to_receive("an order placed event")
        .with_content({
            "orderId": Format().uuid,
            "customerId": Format().uuid,
            "items": [
                {
                    "productId": Format().uuid,
                    "quantity": 1,
                    "unitPrice": 9.99,
                }
            ],
            "totalAmount": 9.99,
            "placedAt": Format().iso_8601_datetime,
        })
        .with_metadata({"contentType": "application/json"})
    )

    with pact:
        handle_order_placed(pact.current_message)

The actual SQS polling/Lambda trigger is tested separately. In a Lambda environment, the handler receives the message body via event["Records"][0]["body"] — that deserialization layer is thin enough that unit testing the handler directly is sufficient for most teams.

Avro and Schema Registry

If you're using Avro with the Confluent Schema Registry, message contracts become more nuanced. The wire format includes a schema ID, not just JSON.

Options:

  1. Test with decoded JSON. Deserialize Avro before the handler. Your contract test uses the decoded JSON shape. This is the most common approach.
  2. Include schema ID in metadata. Some teams put the schema ID in pact metadata and validate it in the provider test.
  3. Use the Schema Registry as a contract. For Avro-heavy stacks, some teams treat schema compatibility checks (FULL_TRANSITIVE compatibility level in the Confluent Schema Registry) as their contract testing strategy instead of Pact.

For most teams starting out, option 1 is the right call. Test the handler, not the wire format.

How Message Contracts Differ from HTTP Pacts

Aspect HTTP Pacts Message Pacts
What's captured Request + Response Message body + metadata
Provider verification Replay HTTP request against running server Call message builder function, compare output
Provider states Set up data before HTTP replay Set up data before builder is called
Transport tested Yes (HTTP method, path, headers) No (topic name, SQS ARN not tested)
Real broker needed No (Pact mock server) No (no transport at all)

The biggest difference: message pacts never touch the transport layer. You're testing the data contract only. The transport (Kafka consumer group, SQS queue URL, SNS subscription) is infrastructure — tested separately or assumed correct.

This is intentional. Pact is about data contracts. If you need to test that your Kafka consumer is subscribed to the right topic with the right consumer group configuration, that's an infrastructure test and belongs in a different layer.

Read more

Start now free