kafka-pipeline

This library provide a robust, easy to use kafka message consumer based on ConsumerGroup from kafka-node that helps you managing message offsets in correct way.

Design notes

The basic idea behind this library is separate consume progress and offset commit into two Transform stream, namely ConsumeStream and CommitStream, messages will be consumed by ConsumeStream by invoking a specified callback within a specified concurrency limit. And, ConsumeStream will yield all received messages in the same order to CommitStream when all the condition following are met:

1. the message itself has been consumed

2. all predecessor message in the same partition have been consumed

While the CommitStream remember the latest offset of each partitions and topics, commit them repeatedly in a specified interval as well as the time when the consumer is rebalanced or closed.

While this idea is originate from ConsumerGroupStream from kafka-node, it's buggy when you try to use it with offset being managed by yourself, which prevents you from implementing a reliable message consumer meets guarantee that each message should be consumed at least once.

Even if you don't care about at-least-once guarantee at all, ConsumerGroupStream duplicates a lot of message after rebalance. Although your apps need to be idempotency when handle message, this library take the best efforts to minimize the duplication of messages, unless error occurred that prevent the offset to be committed, it shall NEVER duplicate any message.

Installation

kafka-node is A peer dependency of this library, you need to install it together with kafka-pipeline.

npm install kafka-pipeline kafka-node

Usage

const {ConsumerGroupPipeline} = require('kafka-pipeline');

const consumerPipeline = new ConsumerGroupPipeline({
  topic: TOPIC_NAME,
  messageConsumer: async (message) => {
    //...
  },
  // A optional callback function handled messages failed to consume by
  // `messageConsumer`. When error raised in `messageConsumer`,
  // consumerPipeline will be closed if you don't provide `failedMessageConsumer`,
  // or `failedMessageConsumer` also raise an exception
  failedMessageConsumer: async (exception, message) => {
    //...
  },
  consumeConcurrency: 8,
  consumeTimeout: 5000,
  commitInterval: 10000,
  consumerGroupOption: {
    groupId: GROUP_ID
    // Properties of this object will be pass to ConsumerGroup
    // except that some of them could be override by ConsumerGroupPipeline
    // See reference of its setting at https://github.com/SOHU-Co/kafka-node
    //
  }
});

consumerPipeline.start();

// The following code is for demo purpose only, not suggesting you to do this
// in production code. But to avoid message duplication you need to ensure your
// app will quit in a graceful way that await the promise returned by
// consumerPipeline.close() is resolved, which will do the final offset commit
process.on('SIGTERM', function () {
  consumerPipeline.close().then(function () {
    process.exit(0);
  });
});

Breaking changes

• 3.0 (Upcoming): consumeConcurrency now limits message consumed concurrently per-partition, not per-consumer, default value changed to 1

• 2.0: Migrated to Typescript

License

ISC License, see LICENSE file.

TODO

1. More test
2. Consider support HighLevelConsumer?