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