Implement Kafka Authentication & clean up timeouts

neighbourhoodie · May 28, 2024 · 9d5d98e · 9d5d98e
1 parent b80b53d
commit 9d5d98e
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 9 deletions.
diff --git a/src/define_config.ts b/src/define_config.ts
@@ -4,8 +4,6 @@ export function defineConfig(config = {}): KafkaConfig {
   return {
     brokers: 'localhost:9092',
     clientId: 'local',
-    connectionTimeout: 3000,
-    requestTimeout: 60000,
     logLevel: 'info',
     // Overwrite default config values if another one is provided
     ...config,

diff --git a/src/index.ts b/src/index.ts
@@ -73,9 +73,13 @@ export class Kafka implements KafkaContract {
   private createKafka() {
     this.#kafka = new KafkaJs({
       brokers: this.getBrokers(),
-      clientId: this.#config.clientId || 'local',
-      connectionTimeout: this.#config.connectionTimeout,
-      requestTimeout: this.#config.requestTimeout,
+      ssl: this.#config.ssl,
+      sasl: this.#config.sasl,
+      clientId: this.#config.clientId,
+      connectionTimeout: this.#config.timeouts?.connection,
+      requestTimeout: this.#config.timeouts?.request,
+      authenticationTimeout: this.#config.timeouts?.authentication,
+      reauthenticationThreshold: this.#config.timeouts?.reauthentication,
       logLevel: toKafkaLogLevel(this.#config.logLevel),
       logCreator: (logLevel: KafkaLogLevel) => {
         this.#logger.level = toAdonisLoggerLevel(logLevel)

diff --git a/src/types.ts b/src/types.ts
@@ -4,7 +4,11 @@ import type {
   ConsumerRunConfig as KafkaConsumerRunConfig,
   Message as KafkaMessage,
   EachMessagePayload as KafkaEachMessagePayload,
+  SASLOptions,
 } from 'kafkajs'
+
+import type tls from 'node:tls'
+
 import type { Level } from '@adonisjs/logger/types'
 import type { Consumer } from './consumer.ts'
 import type { Producer } from './producer.ts'
@@ -37,9 +41,15 @@ declare module '@adonisjs/core/types' {
 
   export interface KafkaConfig {
     brokers?: string | string[]
+    ssl?: tls.ConnectionOptions | boolean
+    sasl?: SASLOptions
     clientId?: string
-    connectionTimeout?: number
-    requestTimeout?: number
+    timeouts?: {
+      connection?: number
+      authentication?: number
+      reauthentication?: number
+      request?: number
+    }
     logLevel: Level
   }
 

diff --git a/stubs/config/kafka.stub b/stubs/config/kafka.stub
@@ -6,8 +6,6 @@ import env from '#start/env'
 const kafkaConfig = {
   brokers: env.get('KAFKA_BROKERS', 'localhost:9092'),
   clientId: env.get('KAFKA_CLIENT_ID'),
-  connectionTimeout: env.get('KAFKA_CONNECTION_TIMEOUT', 3000),
-  requestTimeout: env.get('KAFKA_REQUEST_TIMEOUT', 60000),
   logLevel: env.get('KAFKA_LOG_LEVEL', env.get('LOG_LEVEL')),
 }