Kafka handling non json messages successfully (#291)

* Using proper docker image

* TDD adding test with error case

* Adding safeJsonDeserializer

* Release prepare

* Closing producer

Author: CarlosGamero

docker-compose.yml

@@ -38,7 +38,7 @@ services:
     restart: on-failure
 
   kafka:
-    image: apache/kafka:latest
+    image: apache/kafka:3.7.1
     container_name: kafka
     ports:
       - 9092:9092

packages/kafka/lib/AbstractKafkaConsumer.ts

@@ -12,14 +12,14 @@ import {
   type ConsumerOptions,
   type Message,
   type MessagesStream,
-  jsonDeserializer,
   stringDeserializer,
 } from '@platformatic/kafka'
 import { AbstractKafkaService, type BaseKafkaOptions } from './AbstractKafkaService.ts'
 import { KafkaHandlerContainer } from './handler-container/KafkaHandlerContainer.ts'
 import type { KafkaHandlerRouting } from './handler-container/KafkaHandlerRoutingBuilder.ts'
 import type { KafkaHandler, RequestContext } from './handler-container/index.ts'
 import type { KafkaConfig, KafkaDependencies, TopicConfig } from './types.ts'
+import { safeJsonDeserializer } from './utils/safeJsonDeserializer.js'
 
 export type KafkaConsumerDependencies = KafkaDependencies &
   Pick<QueueConsumerDependencies, 'transactionObservabilityManager'>
@@ -66,7 +66,7 @@ export abstract class AbstractKafkaConsumer<
       autocommit: false, // Handling commits manually
       deserializers: {
         key: stringDeserializer,
-        value: jsonDeserializer,
+        value: safeJsonDeserializer,
         headerKey: stringDeserializer,
         headerValue: stringDeserializer,
       },
@@ -102,6 +102,9 @@ export abstract class AbstractKafkaConsumer<
   }
 
   private async consume(message: Message<string, object, string, string>): Promise<void> {
+    // message.value can be undefined if the message is not JSON-serializable
+    if (!message.value) return message.commit()
+
     const handler = this.handlerContainer.resolveHandler(message.topic, message.value)
     // if there is no handler for the message, we ignore it (simulating subscription)
     if (!handler) return message.commit()

packages/kafka/lib/utils/safeJsonDeserializer.spec.ts

@@ -0,0 +1,38 @@
+import { stringValueSerializer } from '@lokalise/node-core'
+import { safeJsonDeserializer } from './safeJsonDeserializer.js'
+
+describe('safeJsonDeserializer', () => {
+  it('should deserialize valid JSON strings', () => {
+    const validJson = JSON.stringify({ key: 'value' })
+    const result = safeJsonDeserializer(validJson)
+    expect(result).toEqual({ key: 'value' })
+  })
+
+  it('should return undefined for invalid JSON strings', () => {
+    const invalidJson = stringValueSerializer({ key: 'value' })
+    const result = safeJsonDeserializer(invalidJson)
+    expect(result).toBeUndefined()
+  })
+
+  it('should deserialize for valid buffer inputs', () => {
+    const buffer = Buffer.from(JSON.stringify({ key: 'value' }))
+    const result = safeJsonDeserializer(buffer)
+    expect(result).toEqual({ key: 'value' })
+  })
+
+  it('should return undefined for invalid buffer inputs', () => {
+    const invalidBuffer = Buffer.from('invalid json')
+    const result = safeJsonDeserializer(invalidBuffer)
+    expect(result).toBeUndefined()
+  })
+
+  it('should return undefined for invalid inputs', () => {
+    const result = safeJsonDeserializer(1 as any)
+    expect(result).toBeUndefined()
+  })
+
+  it('should return undefined for undefined inputs', () => {
+    const result = safeJsonDeserializer(undefined)
+    expect(result).toBeUndefined()
+  })
+})

packages/kafka/lib/utils/safeJsonDeserializer.ts

@@ -0,0 +1,11 @@
+export const safeJsonDeserializer = (data?: string | Buffer): object | undefined => {
+  if (!data) return undefined
+  // Checking to be safe
+  if (!Buffer.isBuffer(data) && typeof data !== 'string') return undefined
+
+  try {
+    return JSON.parse(data.toString('utf-8'))
+  } catch (_) {
+    return undefined
+  }
+}

packages/kafka/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@message-queue-toolkit/kafka",
-    "version": "0.1.2",
+    "version": "0.1.3",
     "engines": {
         "node": ">= 22.14.0"
     },

packages/kafka/test/consumer/PermissionConsumer.spec.ts

@@ -1,4 +1,6 @@
 import { randomUUID } from 'node:crypto'
+import { waitAndRetry } from '@lokalise/universal-ts-utils/node'
+import { Producer, stringSerializers } from '@platformatic/kafka'
 import { afterAll, expect } from 'vitest'
 import z from 'zod/v3'
 import { KafkaHandlerConfig, type RequestContext } from '../../lib/index.js'
@@ -244,6 +246,31 @@ describe('PermissionConsumer', () => {
       const spy = await consumer.handlerSpy.waitForMessageWithId('1', 'error')
       expect(spy.processingResult).toMatchObject({ errorReason: 'invalidMessage' })
     })
+
+    it('should ignore non json messages', async () => {
+      // Given
+      const errorSpy = vi.spyOn(testContext.cradle.errorReporter, 'report')
+
+      const producer = new Producer<string, string, string, string>({
+        ...testContext.cradle.kafkaConfig,
+        clientId: randomUUID(),
+        autocreateTopics: true,
+        serializers: stringSerializers,
+      })
+      consumer = new PermissionConsumer(testContext.cradle)
+      await consumer.init()
+
+      // When
+      await producer.send({
+        messages: [{ topic: 'permission-general', value: 'not valid json' }],
+      })
+
+      // Then
+      await waitAndRetry(() => errorSpy.mock.calls.length > 0, 10, 100)
+      expect(errorSpy).not.toHaveBeenCalled()
+
+      await producer.close()
+    })
   })
 
   describe('observability - request context', () => {

packages/kafka/test/utils/testContext.ts

@@ -73,6 +73,7 @@ const resolveDIConfig = (awilixManager: AwilixManager): DiConfig => ({
       ({
         report: () => {},
       }) satisfies ErrorReporter,
+    SINGLETON_CONFIG,
   ),
   transactionObservabilityManager: asFunction(
     () =>