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

Perseus · Perseus · commit 8f35fa4cc352 · 2025-05-26T12:41:04.000Z
diff --git a/README.md b/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.
diff --git a/lib/admin.d.ts b/lib/admin.d.ts
@@ -1,6 +1,6 @@
 /// <reference types="node" />
 import Client from './client';
-import { GlobalConfig } from 'node-rdkafka';
+import { GlobalConfig } from '@confluentinc/kafka-javascript';
 import EventEmitter from 'events';
 declare class KafkaAdmin extends Client {
     private adminClient;
diff --git a/lib/admin.js b/lib/admin.js
diff --git a/lib/client.d.ts b/lib/client.d.ts
@@ -1,6 +1,6 @@
 /// <reference types="node" />
 import EventEmitter from "events";
-import { GlobalConfig, LibrdKafkaError, TopicConfig } from "node-rdkafka";
+import { GlobalConfig, LibrdKafkaError, TopicConfig } from "@confluentinc/kafka-javascript";
 export default class Client {
     private clientId;
     private clientType;
diff --git a/lib/consumer.d.ts b/lib/consumer.d.ts
@@ -1,5 +1,5 @@
 /// <reference types="node" />
-import { ConsumerGlobalConfig, ConsumerTopicConfig, LibrdKafkaError, Message, SubscribeTopicList } from 'node-rdkafka';
+import { ConsumerGlobalConfig, ConsumerTopicConfig, LibrdKafkaError, Message, SubscribeTopicList } from '@confluentinc/kafka-javascript';
 import { EventEmitter } from 'stream';
 import Client from './client';
 export declare type ConsumeActionFunction = (err: LibrdKafkaError, messages: Message[]) => void;
@@ -10,8 +10,8 @@ declare 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: string, groupId: string, config: ConsumerGlobalConfig, topicConfig: ConsumerTopicConfig, emitter: EventEmitter);
@@ -24,7 +24,7 @@ declare class KafkaConsumer extends Client {
     connect(): Promise<this | LibrdKafkaError>;
     /**
      * 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: SubscribeTopicList): this;
@@ -62,7 +62,7 @@ declare class KafkaConsumer extends Client {
     _wrapListenCallbackWrapper(actionOnData: any): (msg: any) => void;
     /**
      * 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: any): any;
diff --git a/lib/consumer.js b/lib/consumer.js
diff --git a/lib/index.d.ts b/lib/index.d.ts
@@ -1,7 +1,7 @@
 import getKafkaProducer, { ProduceParameters } from './producer';
 import getKafkaConsumer, { ConsumeActionFunction, ListenActionFunction } from './consumer';
 import KafkaAdmin from './admin';
-import { ClientMetrics, LibrdKafkaError, NumberNullUndefined, SubscribeTopicList } from 'node-rdkafka';
+import { ClientMetrics, LibrdKafkaError, NumberNullUndefined, SubscribeTopicList } from '@confluentinc/kafka-javascript';
 import { ErrorHandlingFunction } from './client';
 interface KafkaConsumer {
     connect(): Promise<KafkaConsumer | LibrdKafkaError>;
diff --git a/lib/producer.d.ts b/lib/producer.d.ts
@@ -1,6 +1,6 @@
 /// <reference types="node" />
 import EventEmitter from 'events';
-import { ClientMetrics, LibrdKafkaError, MessageKey, NumberNullUndefined, ProducerGlobalConfig, ProducerTopicConfig } from 'node-rdkafka';
+import { ClientMetrics, LibrdKafkaError, MessageKey, NumberNullUndefined, ProducerGlobalConfig, ProducerTopicConfig } from '@confluentinc/kafka-javascript';
 import Client, { ErrorHandlingFunction } from './client';
 export interface ProduceParameters {
     topic: string;
@@ -14,8 +14,8 @@ declare 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: string, config: ProducerGlobalConfig, topicConfig: ProducerTopicConfig, emitter: EventEmitter);
@@ -29,17 +29,17 @@ declare 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, key, timestamp }: ProduceParameters): boolean | number;
     /**
      * 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}
      */
diff --git a/lib/producer.js b/lib/producer.js
diff --git a/package.json b/package.json
@@ -18,7 +18,7 @@
   ],
   "license": "ISC",
   "dependencies": {
-    "node-rdkafka": "^2.18.0"
+    "@confluentinc/kafka-javascript": "^1.3.1"
   },
   "devDependencies": {
     "@types/node": "17.0.21",
diff --git a/src/admin.ts b/src/admin.ts
@@ -1,5 +1,5 @@
 import Client from './client';
-import { AdminClient, GlobalConfig } from 'node-rdkafka';
+import { AdminClient, GlobalConfig } from '@confluentinc/kafka-javascript';
 import EventEmitter from 'events';
 
 class KafkaAdmin extends Client {
diff --git a/src/client.ts b/src/client.ts
@@ -1,5 +1,5 @@
 import EventEmitter from "events";
-import { GlobalConfig, LibrdKafkaError, TopicConfig } from "node-rdkafka";
+import { GlobalConfig, LibrdKafkaError, TopicConfig } from "@confluentinc/kafka-javascript";
 
 export default class Client {
     constructor(private clientId: string,
diff --git a/src/consumer.ts b/src/consumer.ts
@@ -1,4 +1,4 @@
-import Kafka, { ConsumerGlobalConfig, ConsumerTopicConfig, LibrdKafkaError, Message, SubscribeTopicList } from 'node-rdkafka';
+import Kafka, { ConsumerGlobalConfig, ConsumerTopicConfig, LibrdKafkaError, Message, SubscribeTopicList } from '@confluentinc/kafka-javascript';
 import { EventEmitter } from 'stream';
 import Client from './client';
 
@@ -15,8 +15,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: string, groupId: string, config: ConsumerGlobalConfig, topicConfig: ConsumerTopicConfig, emitter: EventEmitter) {
@@ -80,7 +80,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: SubscribeTopicList): this {
@@ -185,7 +185,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) {
diff --git a/src/index.ts b/src/index.ts
@@ -1,7 +1,7 @@
 import getKafkaProducer, { ProduceParameters } from './producer'
 import getKafkaConsumer, { ConsumeActionFunction, ListenActionFunction } from './consumer';
 import KafkaAdmin from './admin';
-import { ClientMetrics, LibrdKafkaError, NumberNullUndefined, SubscribeTopicList } from 'node-rdkafka';
+import { ClientMetrics, LibrdKafkaError, NumberNullUndefined, SubscribeTopicList } from '@confluentinc/kafka-javascript';
 import { ErrorHandlingFunction } from './client';
 
 interface KafkaConsumer {
diff --git a/src/producer.ts b/src/producer.ts
@@ -1,5 +1,5 @@
 import EventEmitter from 'events';
-import Kafka, { ClientMetrics, LibrdKafkaError, MessageKey, NumberNullUndefined, ProducerGlobalConfig, ProducerTopicConfig } from 'node-rdkafka';
+import Kafka, { ClientMetrics, LibrdKafkaError, MessageKey, NumberNullUndefined, ProducerGlobalConfig, ProducerTopicConfig } from '@confluentinc/kafka-javascript';
 import Client, { ErrorHandlingFunction } from './client';
 
 export interface ProduceParameters{
@@ -18,8 +18,8 @@ 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: string, config: ProducerGlobalConfig, topicConfig: ProducerTopicConfig, emitter: EventEmitter) {
@@ -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 }: ProduceParameters): boolean | number {
@@ -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}
      */

Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
`1`	`1`	`# kafka-wrapper`
`2`		-A simple kafka wrapper for `node-rdkafka` client.
	`2`	+A simple kafka wrapper for `@confluentinc/kafka-javascript` client.