expect-kafka

A jest extension library for E2E testing of kafka-enabled applications, using Kafkajs

It is recommended to use this library in conjunction with wait-for-expect.

Usage

Basic Usage

const { KafkaTest } = require('expect-kafka');
const { Kafka } = require('kafkajs');
const waitForExpect = require('wait-for-expect');

waitForExpect.defaults.timeout = 10000;
waitForExpect.defaults.interval = 500;

describe('ExpectKafka', () => {
  // Create a kafkajs client
  const kafkaClient = new Kafka({
    brokers: ['localhost:9092'],
  })

  // Create an expect-kafka client
  const expectKafka = new ExpectKafka(kafkaClient);

  it('should send and receive a message', async () => {
    // Subscribe to a topic: 
    // expect-kafka will cache all messages received on this topic
    await expectKafka.subscribe('my-topic', new StringSerializer());

    // Publish a message
    await expectKafka.publish('my-topic', 'my-message');

    // Assert that messages were received by the subscription
    await waitForExpect(() => {

      // call expectKafka.received(topic) to get the received messages as an array
      expect(expectKafka.received('my-topic')).toContainEqual('my-message');
    });
  });
});

Topic Creation

You can use expectKafka.ensureTopics() to create topics before your test suite runs

beforeAll(async () => {
  await expectKafka.ensureTopics([
    'topic01',
    'topic02',
  ], {
    numPartitions: 1,
    replicationFactor: 1
  });
});

Message Serializers

When subscribing to a topic, the serializer determines how the message is stored and returned.

There are 3 built-in message serializers:

• Buffer Serializer: returns messages as raw byte buffers
• String Serializer: returns messages as UTF-8 strings
• Json Serializer: returns messages as Json Objects

await expectKafka.subscribe('my-topic', new BufferSerializer());
expectKafka.received('my-topic'); // Returns an array of Buffers

await expectKafka.subscribe('my-topic', new StringSerializer());
expectKafka.received('my-topic'); // Returns an array of Strings

await expectKafka.subscribe('my-topic', new StringSerializer());
expectKafka.received('my-topic'); // Returns an array of JSON objects

You can pass any object that implements the 'serialize' method

it('custom serializer', async () => {
  // A Custom serializer which parses input as numbers
  const customSerializer: MessageSerializer<number> = {
    serialize: (buffer: Buffer) => parseInt(buffer.toString()),
  };

  await expectKafka.subscribe('custom-serializer', customSerializer);
  await expectKafka.publish('custom-serializer', '1');

  await waitForExpect(() => {
    expect(expectKafka.received('buffer-serializer')).toContainEqual(1);
  })
});

Expect Extensions

Expect-Kafka adds a few expect functions to jest.

toContainObjectMatching

Passes if one object the given array is a partial match to the expectation

const data = [
  {
    one: 1,
    two: 2,
  },
  {
    three: 3,
    four: 4,
  },
]

// Passes because the second element of data is a partial match.
expect(data).toContainObjectMatching({ 
  three: 3
});

Development

Build the container

For portability, e2e tests run in docker

docker build -t expect-kafka .

Run local test suite

Tested on Windows. Docker-compose may require adjustments for mac/linux.

1. Setup local kafka broker with docker-compose up
   • If you tear-down the stack, use docker-compose rm -svf to cleanup ephemeral data
2. Run test suite locally, or in docker as well

   npm run test:local
   or
   docker run -it expect-kafka npm run test:local