kafka-executor

Listens to topics and executes asynchronous functions able to process each kafka message, ensuring that any processing will succeed, before the corresponding message offset is committed.

Installation

Usage

Documentation

Features

• Simple API
• Ensures that all the jobs will be executed successfully before a message will be committed
• Retry strategy for jobs that fail
• Graceful shutdown

Installation

Install with yarn or npm

yarn add kafka-executor
        #or
npm install kafka-executor --save

Usage

Basic

import KafkaExecutor, { Job } from 'kafka-executor';

const executor = new KafkaExecutor({
    brokerList:'0.0.0.0:9092,0.0.0.0:9091',
    topics:['topic1','topic2'],
    groupId:'groupId',
});

executor.init();

executor.addJob('myJobId',new Job((kafkaMessage)=>{
    console.log(kafkaMessage);
    return Promise.resolve();
}));

Documentation

Job

import { Job } from 'kafka-executor';

new Job(() => Promise.resolve(), {
    maxRetries?: number,
    retryDelay?: number | (retryNumber: number) => number,
    shouldRetry?: boolean | (err: Error) => boolean,
})

| Name | Required | Default | Description | | ------------- | ------------- | ------------- | ------------- | | maxRetries: number | no | 3 | How many times must retry until fail | | retryDelay: number | (retryIndex)=>number | no | 60000 ms | The delay between the retries in ms | | shouldRetry: boolean | (error)=>boolean | no | true | Determines if a job have to retry in case of failure |

KafkaExecutor

Options

import KafkaExecutor from 'kafka-executor';

new KafkaExecutor({
    brokerList: string;
    groupId: string;
    topics: string[];
    connectionTimeout: string[];
    checkInterval?: number;
    batchSize?: number;
    errorHandler?: (err: Error[], message:KafkaMessage,commit:Function) => void;
    logger?: (message: string, type: LogType, code?: string) => void;
    maxRetries?: number;
    retryDelay?: number;
    consumer?: object;
})

| Name | Required | Default | Description | | ------------- | ------------- | ------------- | ------------- | | brokerList: string | yes | - | Initial list of brokers as a CSV list of broker host or host:port | | topics: [string] | yes | - | The topics that the consumer will listen to| | groupId: string | yes | - | Client group id string. All clients sharing the same group.id belong to the same group | | checkInterval: number | no | 2000 | How match time to wait until check for new messages in case of dead period | | batchSize: number | no | 1 | How many messages to process concurrently, Change this according to your error tolerance | | errorHandler: (error,kafkaMessage,commit:Function)=>void | no | yes | A function responsible for handling job errors. By Default the process will exit with code 1 | | logger: (message:string, type:'info'|'warn'|'error', code)=>void | no | console | A function responsible for logging | | consumer: object | no | - | Options for the consumer see rdkafka configuration options | | maxRetries: number | no | 3 | Global configuration for all jobs | | retryDelay: number | no | 60000 ms | Global configuration for all jobs |

Functions

import KafkaExecutor from 'kafka-executor';

const executor = new KafkaExecutor({
    brokerList: '0.0.0.0:9092';
    groupId: 'group';
    topics: ['topic'];
});

executor.addJob('myJobId',new Job(...))
executor.init() 
executor.removeJob('myJobId') 
executor.on('event',Function) 
executor.shutdown() 

| Name | Description | | ------------- | ------------- | | init: (jobId:string)=>Promise) | Initialize the kafka-executor and connect consumer with the kafka. | | addJob: (jobId:string, new Job(...))=>void) | Adds a job in the processing flow. | | removeJob: (jobId:string)=>void) | removes a job. | | on: (jobId:string)=>void) | Listens in a variant of events handled by kafka-executor and rdkafka | | shutdown: (jobId:string)=>Promise) | shutdown the process gracefully ensuring that the pending jobs will finish before exit |

Events

| Event | Arguments | Description | | ------------- | ------------- | ------------- | | message.received | kafkaMessage[] | Fires when the consumer gets a message | | message.committed | kafkaMessage | Fires when the consumer commits a message | | processing.error | kafkaMessage, error | Fires when one or more jobs fail | | shutdown | - | Fires when the kafka-executor shutdown |

node-rdkafka events

|Event|Description| |-------|----------| |data | When using the Standard API consumed messages are emitted in this event. | |disconnected | The disconnected event is emitted when the broker disconnects. This event is only emitted when .disconnect is called. The wrapper will always try to reconnect otherwise. | |ready | The ready event is emitted when the Consumer is ready to read messages. | |event | The event event is emitted when librdkafka reports an event (if you opted in via the event_cb option).| |event.log | The event.log event is emitted when logging events occur (if you opted in for logging via the event_cb option). You will need to set a value for debug if you want information to send. | |event.stats | The event.stats event is emitted when librdkafka reports stats (if you opted in by setting the statistics.interval.ms to a non-zero value). | |event.throttle | The event.throttle event is emitted when librdkafka reports throttling.|

kafkaMessage

{
    value: Buffer, 
    size: number,
    topic: string, 
    offset: number, 
    partition: number, 
    key: string, 
    timestamp: number
}

| Name | Type | Description | | ------------- | ------------- | ------------- | | value | Buffer | message contents as a Buffer | | size | number | size of the message, in bytes | | topic | string | topic the message comes from | | offset | number | offset the message was read from | | partition | string | partition the message was on | | key | number | key of the message if present | | timestamp | number | timestamp of message creation |

error

{
    ...Error,
    jobId: string,
    status?: string,
}

| Name | type | Description | | ------------- | ------------- | ------------- | | jobId | the failed job | | status | the http status if exists |

codes

| Name | Description | | ------------- | ------------- | | kafkaError | Log produced by kafka | | connectionError | Log produced when trying to connect to kafka | | jobFailed | Log produced by a job | | jobRetry | Log produced by a job when retries|