kafka-node-reply
A kafka-node does not support synchronouse send message and wait for response message from another topic. So, this package may can help you do that and make it simpler. You can use it instead HTTP protocol
Let's getting started! Just install this package and use it as a library
Requirement
- Kafka node
kafka-node
version4.0.0
Kafka-node is a Node.js client for Apache Kafka 0.9 and later.
Installation
- Install package
kafka-node-reply
is super easy to install.
npm install kafka-node-reply
API
Concept
Each instance of kafkaNodeReply will have two part. The first is producer will send a message request to topic requestTopic
with a formatted message was defined (formatMessage
). Other consumers consume this topic will receive and process base on message request. After the process is done will send response into replyTopic
that kafkaNodeReply current consume to receive message response and complete this request
KafkaNodeReply
Create new instance of kafkaNodeReply.
client
The kafka-node client. Can create new client or re-use from kafka-node
admin
The kafka-node admin. Can create new admin or re-use from kafka-node client
requestTopicOptions
The options for producer
responseTopicOptions
The options for response topic that kafkaNodeReply will consumer to receive response message
options
The options you are self define generic. It can merge from requestTopicOptions.options, responseTopicOptions.options, kafka-node client options...
requestSync
Sent a request in synchronuos
message
The message request to communicate. All of kafkaNodeRpely instances must use this format message. You can imagine this message similar http request, contain info such as: body, headers, params...
How to use
Example
This package using base on kafka-node
. So you can re-use theirs client to connect this package. Such as: kafka-node.KafkaClient
, kafka-node.Admin
const kafka = ;const kafkaReply = ; let client = options;let admin = client; let requestTopicTops = "topic": "TopicRequest" "requestTimeout": 30000 "options": "partitionerType": 2 "requireAcks": 0 let responseTopicOptions = "topic": "TopicReply" "options": "groupId": "group-A-reply" // Create new kafka request response instancelet kafkaReqRes = client admin requestTopicTops responseTopicOptions options let message = "ation": "getUser" "body": "userId": 1 "email": "example@email.com" "headers": "ContentType": "json/application" kafkaReqRes
KafkaNodeReply
TODO
- Solution for kafka consumer group rebalancing make consumer cannot receive message
- Support request-reply stream
- Control error crash consumer
Change log
1.0.1
- Catch error when client disconnect
1.0.0
Release 1.0.0
Author
Quyen Hung - hungquyen.ntt@gmail.com