(chore): migrate to using confluent's kafka library

Author: Perseus

README.md

@@ -1,2 +1,2 @@
 # kafka-wrapper
-A simple kafka wrapper for `node-rdkafka` client.
+A simple kafka wrapper for `@confluentinc/kafka-javascript` client.

package.json

@@ -18,6 +18,7 @@
   ],
   "license": "ISC",
   "dependencies": {
+    "@confluentinc/kafka-javascript": "^1.3.1",
     "node-rdkafka": "^2.12.0"
   }
 }

src/admin.js

@@ -1,5 +1,5 @@
 const Client = require('./client');
-const Kafka = require('node-rdkafka');
+const Kafka = require('@confluentinc/kafka-javascript');
 
 class KafkaAdmin extends Client {
 

src/consumer.js

@@ -1,4 +1,4 @@
-const Kafka = require('node-rdkafka');
+const Kafka = require('@confluentinc/kafka-javascript');
 const Client = require('./client');
 
 class KafkaConsumer extends Client {
@@ -7,8 +7,8 @@ class KafkaConsumer extends Client {
      * Initializes a KafkaConsumer.
      * @param {String} clientId: id to identify a client consuming the message. 
      * @param {String} groupId: consumer group id, the consumer belongs to. 
-     * @param {import('node-rdkafka').ConsumerGlobalConfig} config: configs for consumer.
-     * @param {import('node-rdkafka').ConsumerTopicConfig} topicConfig: topic configs 
+     * @param {import('@confluentinc/kafka-javascript').ConsumerGlobalConfig} config: configs for consumer.
+     * @param {import('@confluentinc/kafka-javascript').ConsumerTopicConfig} topicConfig: topic configs 
      * @param {EventEmitter} emitter: to emit log events
      */
     constructor(clientId, groupId, config, topicConfig, emitter) {
@@ -74,7 +74,7 @@ class KafkaConsumer extends Client {
 
     /**
      * Subscribe to topics.
-     * @param {import('node-rdkafka').SubscribeTopicList} topics: array of topic names. 
+     * @param {import('@confluentinc/kafka-javascript').SubscribeTopicList} topics: array of topic names. 
      * @returns {KafkaConsumer}
      */
     subscribe(topics) {
@@ -179,7 +179,7 @@ class KafkaConsumer extends Client {
 
     /**
      * Parses message before passing it to consumer callback.
-     * @param {Object} msg - expects it to be in node-rdkafka msg format. 
+     * @param {Object} msg - expects it to be in @confluentinc/kafka-javascript msg format. 
      * @returns 
      */
     _parseMessage(msg) {

src/producer.js

@@ -1,13 +1,13 @@
-const Kafka = require('node-rdkafka');
+const Kafka = require('@confluentinc/kafka-javascript');
 const Client = require('./client');
 
 class KafkaProducer extends Client {
 
     /**
      * Initializes a KafkaProducer.
      * @param {String} clientId: id to identify a client producing the message.
-     * @param {import('node-rdkafka').ProducerGlobalConfig} config: configs for producer.
-     * @param {import('node-rdkafka').ProducerTopicConfig} topicConfig: topic configs.
+     * @param {import('@confluentinc/kafka-javascript').ProducerGlobalConfig} config: configs for producer.
+     * @param {import('@confluentinc/kafka-javascript').ProducerTopicConfig} topicConfig: topic configs.
      * @param {EventEmitter} emitter: to emit log messages
      */
     constructor(clientId, config, topicConfig, emitter) {
@@ -85,10 +85,10 @@ class KafkaProducer extends Client {
     /**
      * Produce a message to a topic-partition.
      * @param {String} topic: name of topic 
-     * @param {import('node-rdkafka').NumberNullUndefined} partition: partition number to produce to.
+     * @param {import('@confluentinc/kafka-javascript').NumberNullUndefined} partition: partition number to produce to.
      * @param {any} message: message to be produced. 
-     * @param {import('node-rdkafka').MessageKey} key: key associated with the message.
-     * @param {import('node-rdkafka').NumberNullUndefined} timestamp: timestamp to send with the message. 
+     * @param {import('@confluentinc/kafka-javascript').MessageKey} key: key associated with the message.
+     * @param {import('@confluentinc/kafka-javascript').NumberNullUndefined} timestamp: timestamp to send with the message. 
      * @returns {import('../types').BooleanOrNumber}: returns boolean or librdkafka error code.
      */
     produce({ topic, message, partition = null, key = null, timestamp = null }) {
@@ -105,7 +105,7 @@ class KafkaProducer extends Client {
     /**
      * Flush everything on the internal librdkafka buffer. 
      * Good to perform before disconnect.
-     * @param {import('node-rdkafka').NumberNullUndefined}} timeout 
+     * @param {import('@confluentinc/kafka-javascript').NumberNullUndefined}} timeout 
      * @param {import('../types').ErrorHandlingFunction} postFlushAction 
      * @returns {KafkaProducer}
      */

types/index.d.ts

@@ -1,4 +1,4 @@
-import { ClientMetrics, ConsumerGlobalConfig, ConsumerTopicConfig, GlobalConfig, LibrdKafkaError, Message, MessageKey, NewTopic, NumberNullUndefined, ProducerGlobalConfig, ProducerTopicConfig, SubscribeTopicList } from "node-rdkafka";
+import { ClientMetrics, ConsumerGlobalConfig, ConsumerTopicConfig, GlobalConfig, LibrdKafkaError, Message, MessageKey, NewTopic, NumberNullUndefined, ProducerGlobalConfig, ProducerTopicConfig, SubscribeTopicList } from "@confluentinc/kafka-javascript";
 
 export type ConsumeActionFunction = (err: LibrdKafkaError, messages: Message[]) => void;