Feat: Kafka publisher connection check on init (#286)

* Improving imports

* Implementing publisher connection check

* Typo

* Improving init

* Adding tests

* Adding comment

Author: CarlosGamero

packages/kafka/lib/AbstractKafkaPublisher.ts

@@ -35,10 +35,12 @@ export abstract class AbstractKafkaPublisher<
     MessageSchemaContainer<SupportedMessageValuesInput<TopicsConfig>>
   >
 
-  private producer?: Producer<string, object, string, object>
+  private readonly producer: Producer<string, object, string, object>
+  private isInitiated: boolean
 
   constructor(dependencies: KafkaDependencies, options: KafkaPublisherOptions<TopicsConfig>) {
     super(dependencies, options)
+    this.isInitiated = false
 
     this.topicsConfig = options.topicsConfig
     if (this.topicsConfig.length === 0) throw new Error('At least one topic must be defined')
@@ -51,10 +53,6 @@ export abstract class AbstractKafkaPublisher<
         messageDefinitions: [],
       })
     }
-  }
-
-  init(): Promise<void> {
-    if (this.producer) return Promise.resolve()
 
     this.producer = new Producer({
       ...this.options.kafka,
@@ -66,13 +64,28 @@ export abstract class AbstractKafkaPublisher<
         headerValue: jsonSerializer,
       },
     })
+  }
 
-    return Promise.resolve()
+  async init(): Promise<void> {
+    if (this.isInitiated) return
+
+    try {
+      await this.producer.listApis()
+      this.isInitiated = true
+    } catch (e) {
+      throw new InternalError({
+        message: 'Producer init failed',
+        errorCode: 'KAFKA_PRODUCER_INIT_ERROR',
+        cause: e,
+      })
+    }
   }
 
   async close(): Promise<void> {
-    await this.producer?.close()
-    this.producer = undefined
+    if (!this.isInitiated) return
+
+    await this.producer.close()
+    this.isInitiated = false
   }
 
   async publish<Topic extends SupportedTopics<TopicsConfig>>(

packages/kafka/lib/AbstractKafkaService.ts

@@ -5,8 +5,9 @@ import {
   resolveGlobalErrorLogObject,
   stringValueSerializer,
 } from '@lokalise/node-core'
-import type { HandlerSpy, HandlerSpyParams } from '@message-queue-toolkit/core'
 import {
+  type HandlerSpy,
+  type HandlerSpyParams,
   type MessageProcessingResult,
   type PublicHandlerSpy,
   resolveHandlerSpy,

packages/kafka/test/publisher/PermissionPublisher.spec.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from 'node:crypto'
 import { InternalError } from '@lokalise/node-core'
 import {
   PERMISSION_ADDED_SCHEMA,
@@ -52,6 +53,26 @@ describe('PermissionPublisher - init', () => {
       )
     })
 
+    it('should fail if kafka is not available', async () => {
+      // Given
+      publisher = new PermissionPublisher(testContext.cradle, {
+        kafka: { clientId: randomUUID(), bootstrapBrokers: ['localhost:9090'] },
+      })
+
+      // When - Then
+      await expect(publisher.init()).rejects.toThrowErrorMatchingInlineSnapshot(
+        '[InternalError: Producer init failed]',
+      )
+    })
+
+    it('should not fail on close if publisher is not started ', async () => {
+      // Given
+      publisher = new PermissionPublisher(testContext.cradle)
+
+      // When - Then
+      await expect(publisher.close()).resolves.not.toThrow()
+    })
+
     it('should fail if topic does not exists', async () => {
       // Given
       publisher = new PermissionPublisher(testContext.cradle, {