Typescript types,

configuration mapping,
return RecordMetadata array

Author: emasab

index.d.ts

@@ -7,10 +7,12 @@ import {
     ConsumerTopicConfig,
     ProducerGlobalConfig,
     ProducerTopicConfig,
-} from './config';
+} from './types/config';
 
-export * from './config';
-export * from './errors';
+export * from './types/config';
+export * from './types/errors';
+import { Kafka } from './types/kafkajs';
+import * as errors from './types/errors';
 
 export interface LibrdKafkaError {
     message: string;
@@ -345,3 +347,21 @@ export interface IAdminClient {
 export abstract class AdminClient {
     static create(conf: GlobalConfig): IAdminClient;
 }
+
+export type RdKafka = {
+  Consumer: KafkaConsumer,
+  Producer: Producer,
+  HighLevelProducer: HighLevelProducer,
+  AdminClient: AdminClient,
+  KafkaConsumer: KafkaConsumer,
+  createReadStream: typeof KafkaConsumer.createReadStream,
+  createWriteStream: typeof Producer.createWriteStream,
+  CODES: typeof errors.CODES,
+  Topic: (name: string) => string,
+  features: typeof features,
+  librdkafkaVersion: typeof librdkafkaVersion,
+}
+
+export type KafkaJS = {
+  Kafka: Kafka
+}

lib/kafkajs/_kafka.js

@@ -15,7 +15,37 @@ class Producer {
     #state = ProducerState.INIT;
 
     constructor(config) {
-        this.#config = config;
+        this.#config = this.#kafkaJSToRdKafkaConfig(config);
+    }
+
+    #kafkaJSToRdKafkaConfig(config) {
+      const ret = {
+        'dr_cb': 'true',
+        'allow.auto.create.topics': 'false'
+      }
+      ret['bootstrap.servers'] = config['brokers'].join(',');
+
+      let withSASL = false;
+
+      if (config.sasl) {
+        const sasl  = config.sasl;
+        if (sasl.mechanism === 'plain' &&
+            typeof sasl.username === 'string' &&
+            typeof sasl.password === 'string') {
+            ret['sasl.mechanism'] = 'PLAIN';
+            ret['sasl.username'] = sasl.username;
+            ret['sasl.password'] = sasl.password;
+            withSASL = true;
+        }
+      }
+
+      if (config.ssl === true && withSASL) {
+        ret['security.protocol'] = 'sasl_ssl';
+      } else if (withSASL) {
+        ret['security.protocol'] = 'sasl_plaintext';
+      }
+
+      return ret;
     }
 
     #readyCb(arg) {
@@ -48,7 +78,17 @@ class Producer {
             }
             //console.log('delivery-report: ' + JSON.stringify(report));
             delete report['opaque'];
-            opaque.resolve(report);
+
+            const recordMetadata = {
+              topicName: report.topic,
+              partition: report.partition,
+              errorCode: 0,
+              baseOffset: report.offset,
+              logAppendTime: null,
+              logStartOffset: null,
+            }
+
+            opaque.resolve(recordMetadata);
         });
 
         // Resolve the promise.
@@ -77,7 +117,6 @@ class Producer {
 
         this.#internalClient.on('disconnected', (arg) => {
             this.#state = ProducerState.DISCONNECTED;
-            console.log('producer disconnected. ' + JSON.stringify(arg));
         });
 
         return new Promise((resolve, reject) => {
@@ -96,15 +135,7 @@ class Producer {
       this.#internalClient.disconnect();
     }
 
-    // producer.send({
-    //     topic: <String>,
-    //     messages: <Message[]>,
-    //     acks: <Number>,
-    //     timeout: <Number>,
-    //     compression: <CompressionTypes>,
-    // })
-
-    send(sendOptions) {
+    async send(sendOptions) {
         if (this.#state !== ProducerState.CONNECTED) {
             return Promise.reject("Cannot send message without awaiting connect()");
         }
@@ -138,7 +169,34 @@ class Producer {
             }));
 
         }
-        return Promise.all(msgPromises);
+        const recordMetadataArr = await Promise.all(msgPromises);
+
+        const topicPartitionRecordMetadata = new Map();
+        for (const recordMetadata of recordMetadataArr) {
+          const key = `${recordMetadata.topicName},${recordMetadata.partition}`;
+          if (recordMetadata.baseOffset == null || !topicPartitionRecordMetadata.has(key)) {
+            topicPartitionRecordMetadata.set(key, recordMetadata);
+            continue;
+          }
+
+          const currentRecordMetadata = topicPartitionRecordMetadata.get(key);
+
+          // Don't overwrite a null baseOffset
+          if (currentRecordMetadata.baseOffset == null) {
+            continue;
+          }
+
+          if (currentRecordMetadata.baseOffset > recordMetadata.baseOffset) {
+            topicPartitionRecordMetadata.set(key, recordMetadata);
+          }
+        }
+
+        const ret = [];
+        for (const [key, value] of topicPartitionRecordMetadata.entries()) {
+          value.baseOffset = value.baseOffset?.toString();
+          ret.push(value);
+        }
+        return ret;
     }
 }
 

package-lock.json

@@ -4,12 +4,14 @@
   "description": "Node.js bindings for librdkafka",
   "librdkafka": "2.2.0",
   "main": "lib/index.js",
+  "types": "types/index.d.ts",
   "scripts": {
     "configure": "node-gyp configure",
     "build": "node-gyp build",
     "test": "make test",
     "install": "node-pre-gyp install --fallback-to-build",
-    "prepack": "node ./ci/prepublish.js"
+    "prepack": "node ./ci/prepublish.js",
+    "test:types": "tsc -p ."
   },
   "binary": {
     "module_name": "node-librdkafka",
@@ -38,19 +40,21 @@
   ],
   "license": "MIT",
   "devDependencies": {
+    "@types/node": "^20.4.5",
     "bluebird": "^3.5.3",
     "jsdoc": "^3.4.0",
     "jshint": "^2.10.1",
     "mocha": "^10.2.0",
     "node-gyp": "^9.3.1",
-    "toolkit-jsdoc": "^1.0.0"
+    "toolkit-jsdoc": "^1.0.0",
+    "typescript": "^5.1.6"
   },
   "dependencies": {
     "@mapbox/node-pre-gyp": "^1.0.11",
     "bindings": "^1.3.1",
     "nan": "^2.17.0"
   },
   "engines": {
-    "node": ">=6.0.0"
+    "node": ">=14.0.0"
   }
 }

package.json

@@ -0,0 +1,18 @@
+{
+  "compilerOptions": {
+    "module": "commonjs",
+    "lib": ["es6"],
+    "noImplicitAny": true,
+    "noImplicitThis": true,
+    "strictNullChecks": true,
+    "baseUrl": ".",
+    "types": ["node_modules/@types/node"],
+    "typeRoots": ["."],
+    "noEmit": true,
+    "forceConsistentCasingInFileNames": true,
+    "strictFunctionTypes": true
+  },
+  "files": [
+    "index.d.ts"
+  ]
+}

tsconfig.json

@@ -0,0 +1,101 @@
+import * as tls from 'tls'
+
+export type BrokersFunction = () => string[] | Promise<string[]>
+
+export type Mechanism = {
+  mechanism: string
+}
+
+type SASLMechanismOptionsMap = {
+  plain: { username: string; password: string }
+  'scram-sha-256': { username: string; password: string }
+  'scram-sha-512': { username: string; password: string }
+}
+
+export type SASLMechanism = keyof SASLMechanismOptionsMap
+type SASLMechanismOptions<T> = T extends SASLMechanism
+  ? { mechanism: T } & SASLMechanismOptionsMap[T]
+  : never
+export type SASLOptions = SASLMechanismOptions<SASLMechanism>
+
+export interface KafkaConfig {
+  brokers: string[] | BrokersFunction
+  ssl?: boolean
+  sasl?: SASLOptions | Mechanism
+  clientId?: string
+  connectionTimeout?: number
+  authenticationTimeout?: number
+  reauthenticationThreshold?: number
+  requestTimeout?: number
+  enforceRequestTimeout?: boolean
+}
+
+export interface ProducerConfig {
+  metadataMaxAge?: number
+  allowAutoTopicCreation?: boolean
+  idempotent?: boolean
+  transactionalId?: string
+  transactionTimeout?: number
+  maxInFlightRequests?: number
+}
+
+export interface IHeaders {
+  [key: string]: Buffer | string | (Buffer | string)[] | undefined
+}
+
+export interface Message {
+  key?: Buffer | string | null
+  value: Buffer | string | null
+  partition?: number
+  headers?: IHeaders
+  timestamp?: string
+}
+
+export enum CompressionTypes {
+  None = 0,
+  GZIP = 1,
+  Snappy = 2,
+  LZ4 = 3,
+  ZSTD = 4,
+}
+
+export var CompressionCodecs: {
+  [CompressionTypes.GZIP]: () => any
+  [CompressionTypes.Snappy]: () => any
+  [CompressionTypes.LZ4]: () => any
+  [CompressionTypes.ZSTD]: () => any
+}
+
+export interface ProducerRecord {
+  topic: string
+  messages: Message[]
+  acks?: number
+  timeout?: number
+  compression?: CompressionTypes
+}
+
+export type RecordMetadata = {
+  topicName: string
+  partition: number
+  errorCode: number
+  offset?: string
+  timestamp?: string
+  baseOffset?: string
+  logAppendTime?: string
+  logStartOffset?: string
+}
+
+export class Kafka {
+  constructor(config: KafkaConfig)
+  producer(config?: ProducerConfig): Producer
+}
+
+type Sender = {
+  send(record: ProducerRecord): Promise<RecordMetadata[]>
+}
+
+export type Producer = Sender & {
+  connect(): Promise<void>
+  disconnect(): Promise<void>
+}
+