API Testing

Contract Testing for Event-Driven Systems with AsyncAPI

HelpMeTest

HelpMeTest

— 4 min read
Contract Testing for Event-Driven Systems with AsyncAPI

REST APIs have OpenAPI. Event-driven systems — Kafka topics, RabbitMQ queues, WebSocket channels — have AsyncAPI.

If your microservices communicate via events and you're not validating message schemas, you're flying blind. A producer changes a field name, drops a required field, or changes a type. Every downstream consumer breaks silently. No compile error, no deployment failure — just missing data and broken behavior that surfaces hours later in logs no one was watching.

AsyncAPI is the spec format for event-driven APIs. This post covers what it looks like, how to validate messages against it, and how to build a contract testing layer for async systems.

What AsyncAPI Looks Like

AsyncAPI spec looks familiar if you've used OpenAPI. Here's a basic spec for a user service that publishes events to Kafka:

asyncapi: '2.6.0'
info:
  title: User Service Events
  version: '1.0.0'

servers:
  production:
    url: kafka.example.com:9092
    protocol: kafka

channels:
  user.created:
    description: Published when a new user registers
    subscribe:
      operationId: receiveUserCreated
      message:
        $ref: '#/components/messages/UserCreated'

  user.deleted:
    description: Published when a user account is deleted
    subscribe:
      operationId: receiveUserDeleted
      message:
        $ref: '#/components/messages/UserDeleted'

components:
  messages:
    UserCreated:
      name: UserCreated
      payload:
        type: object
        required: [id, email, createdAt]
        properties:
          id:
            type: string
            format: uuid
          email:
            type: string
            format: email
          name:
            type: string
          createdAt:
            type: string
            format: date-time
    UserDeleted:
      name: UserDeleted
      payload:
        type: object
        required: [id, deletedAt]
        properties:
          id:
            type: string
            format: uuid
          deletedAt:
            type: string
            format: date-time

This spec documents what messages look like, on which channels, and what direction (publish vs subscribe). It's the contract between services.

Validating Messages Against the Spec

The @asyncapi/parser package parses specs and gives you schema objects you can validate against:

const { parse } = require('@asyncapi/parser');
const Ajv = require('ajv');
const addFormats = require('ajv-formats');
const fs = require('fs');
const yaml = require('js-yaml');

async function loadSchemas() {
  const specContent = fs.readFileSync('./asyncapi.yaml', 'utf8');
  const doc = await parse(specContent);
  
  const schemas = {};
  for (const [channelName, channel] of Object.entries(doc.channels())) {
    const message = channel.subscribe()?.message() || channel.publish()?.message();
    if (message) {
      schemas[channelName] = message.payload().json();
    }
  }
  return schemas;
}

async function createValidator() {
  const schemas = await loadSchemas();
  const ajv = new Ajv({ allErrors: true });
  addFormats(ajv);
  
  return {
    validate(channel, message) {
      const schema = schemas[channel];
      if (!schema) throw new Error(`Unknown channel: ${channel}`);
      
      const validate = ajv.compile(schema);
      const valid = validate(message);
      
      if (!valid) {
        throw new Error(
          `Message on ${channel} violates schema:\n` +
          ajv.errorsText(validate.errors)
        );
      }
      return true;
    }
  };
}

Use this validator in your consumer before processing messages:

const validator = await createValidator();

consumer.on('message', (channel, rawMessage) => {
  const message = JSON.parse(rawMessage.value.toString());
  
  try {
    validator.validate(channel, message);
    processMessage(channel, message);
  } catch (err) {
    console.error('Schema violation received:', err.message);
    // Dead letter queue, alert, etc.
  }
});

Contract Testing Producers

The producer contract test verifies that what your service publishes matches the spec. Write this as a unit test that runs in CI:

const { createValidator } = require('./asyncapi-validator');
const UserService = require('../services/UserService');

describe('UserService event contracts', () => {
  let validator;
  
  beforeAll(async () => {
    validator = await createValidator();
  });

  test('createUser publishes a valid UserCreated event', async () => {
    const published = [];
    
    // Mock the Kafka producer to capture what gets published
    const mockProducer = {
      send: async ({ topic, messages }) => {
        published.push({ topic, message: JSON.parse(messages[0].value) });
      }
    };
    
    const service = new UserService({ producer: mockProducer });
    await service.createUser({ name: 'Alice', email: 'alice@example.com' });
    
    expect(published).toHaveLength(1);
    expect(published[0].topic).toBe('user.created');
    
    // This throws if the message doesn't match the spec
    expect(() => validator.validate('user.created', published[0].message)).not.toThrow();
  });
});

Contract Testing Consumers

The consumer contract test verifies that your service can handle the full range of valid messages:

describe('UserCreated consumer', () => {
  const validUserCreatedMessage = {
    id: '123e4567-e89b-12d3-a456-426614174000',
    email: 'alice@example.com',
    name: 'Alice',
    createdAt: '2024-01-15T10:30:00Z'
  };

  test('processes a valid UserCreated message', async () => {
    const consumer = new UserCreatedConsumer({ db: mockDb });
    await consumer.handle(validUserCreatedMessage);
    
    expect(mockDb.users.insert).toHaveBeenCalledWith(
      expect.objectContaining({ email: 'alice@example.com' })
    );
  });

  test('handles missing optional name field', async () => {
    const messageWithoutName = { ...validUserCreatedMessage };
    delete messageWithoutName.name;
    
    // name is optional in spec — consumer must handle this
    const consumer = new UserCreatedConsumer({ db: mockDb });
    await expect(consumer.handle(messageWithoutName)).resolves.not.toThrow();
  });
});

Integration Testing with Testcontainers

For integration tests, spin up real Kafka:

const { KafkaContainer } = require('@testcontainers/kafka');
const { Kafka } = require('kafkajs');

describe('UserService Kafka integration', () => {
  let kafkaContainer;
  let kafka;
  let validator;

  beforeAll(async () => {
    kafkaContainer = await new KafkaContainer().start();
    kafka = new Kafka({
      clientId: 'test',
      brokers: [`${kafkaContainer.getHost()}:${kafkaContainer.getMappedPort(9093)}`]
    });
    validator = await createValidator();
  }, 60000);

  afterAll(async () => {
    await kafkaContainer.stop();
  });

  test('published messages match AsyncAPI contract', async () => {
    const consumer = kafka.consumer({ groupId: 'test-group' });
    await consumer.connect();
    await consumer.subscribe({ topic: 'user.created' });

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

    // Trigger the producer
    const service = new UserService({ kafka });
    await service.createUser({ name: 'Bob', email: 'bob@example.com' });

    await new Promise(r => setTimeout(r, 2000));

    expect(received).toHaveLength(1);
    validator.validate('user.created', received[0]); // throws if invalid
  });
});

Schema Registry Integration

If you're using Confluent Schema Registry or AWS Glue with Avro/Protobuf schemas, you can cross-reference your AsyncAPI spec against the registry to keep them in sync:

const { SchemaRegistry } = require('@kafkajs/confluent-schema-registry');

async function verifyRegistryMatchesSpec() {
  const registry = new SchemaRegistry({ host: 'http://schema-registry:8081' });
  const schemas = await loadSchemas(); // from AsyncAPI spec
  
  for (const [channel, specSchema] of Object.entries(schemas)) {
    const topicName = channelToTopic(channel);
    const registrySchema = await registry.getLatestSchemaId(`${topicName}-value`);
    const registryDef = await registry.getSchema(registrySchema);
    
    // Compare — divergence means spec and registry are out of sync
    assertSchemasCompatible(specSchema, registryDef);
  }
}

Keeping Consumers Safe

The production risk with event-driven systems is a producer deploying a breaking change. Your consumers fail silently or crash. The patterns that protect you:

  1. Validate on consume — reject messages that violate the spec, route to dead letter queues
  2. Consumer contract tests in CI — catch incompatibilities before deploy
  3. Schema evolution rules — only additive changes (new optional fields), never remove required fields
  4. Version channels — user.created.v2 instead of breaking user.created

Async contracts are harder to enforce than REST because there's no synchronous error response. The producer never sees that consumers are failing. This makes upfront schema discipline more important, not less.

For continuous visibility into whether your consumers are processing events successfully in production, HelpMeTest lets you write end-to-end monitoring scenarios that verify the full event flow — from trigger to downstream side effect — running 24/7 and alerting you when event processing breaks.

Read more

Start now free