Confluentinc kafka javascript (#279)

* Adding @confluentinc/kafka-javascript

* Trying to add test

* Fixing threads issue

* Minor change

* Adding kafka-ui

* Fixing name in kafka-ui

* Fixing test

* Adding kafka ci for node 20

* Simplifying CI action

* Adding back platformatic

* Kafka requires node 22.14

* Separating tests

* Adding Kafka CI back

* Typo fix

* Removing kaka CI for node 24 for now

Author: CarlosGamero

.github/workflows/ci.common.yml

@@ -26,7 +26,7 @@ jobs:
       # We need to trigger the rebuild of @confluentinc/kafka-javascript to run scrips
       - name: Run @confluentic/kafka-javascript scripts
         if: ${{ inputs.package_name == '@message-queue-toolkit/kafka' }}
-        run: npm rebuild rebuild @confluentinc/kafka-javascript
+        run: npm rebuild @confluentinc/kafka-javascript
 
       - name: Build TS
         run: npm run build -- --filter=${{ inputs.package_name }}

.github/workflows/ci.yml

@@ -27,6 +27,15 @@ jobs:
       node_version: ${{ matrix.node-version }}
       package_name: ${{ matrix.package-name }}
 
+  kafka:
+    strategy:
+      matrix:
+        node-version: [22.x]
+    uses: kibertoad/message-queue-toolkit/.github/workflows/ci.common.yml@main
+    with:
+      package_name: '@message-queue-toolkit/kafka'
+      node_version: ${{ matrix.node-version }}
+
   automerge:
     needs: [ general ]
     runs-on: ubuntu-latest

docker-compose.yml

@@ -42,6 +42,33 @@ services:
     container_name: kafka
     ports:
       - 9092:9092
+    environment:
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_LISTENERS: LOCAL://0.0.0.0:9092,DOCKER://kafka:9093,CONTROLLER://localhost:9094
+      KAFKA_ADVERTISED_LISTENERS: LOCAL://localhost:9092,DOCKER://kafka:9093
+      KAFKA_INTER_BROKER_LISTENER_NAME: LOCAL
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,LOCAL:PLAINTEXT,DOCKER:PLAINTEXT
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9094
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_NUM_PARTITIONS: 1
+    restart: on-failure
+
+  kafka-ui:
+    container_name: kafka-ui
+    image: provectuslabs/kafka-ui:latest
+    ports:
+      - 8080:8080
+    depends_on:
+      - kafka
+    environment:
+      DYNAMIC_CONFIG_ENABLED: 'true'
+      KAFKA_CLUSTERS_0_NAME: Local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9093
     restart: on-failure
 
 volumes:

packages/kafka/lib/test-confluentic.spec.ts

@@ -0,0 +1,73 @@
+import { randomUUID } from 'node:crypto'
+import { KafkaJS } from '@confluentinc/kafka-javascript'
+import { waitAndRetry } from '@lokalise/universal-ts-utils/node'
+import {} from '@platformatic/kafka'
+import { type TestContext, registerDependencies } from '../test/testContext.ts'
+
+describe('Test confluentic-kafka', () => {
+  let testContext: TestContext
+
+  beforeAll(async () => {
+    testContext = await registerDependencies()
+  })
+
+  afterAll(async () => {
+    await testContext.dispose()
+  })
+
+  it('should send and receive a message', async () => {
+    // Given
+    const clientId = randomUUID()
+    const groupId = randomUUID()
+    // Use a fresh, unique topic per run to avoid stale state
+    const topic = `test-topic-${Date.now()}`
+    const messageValue = 'My test message'
+
+    const kafka = new KafkaJS.Kafka({
+      'client.id': clientId,
+      'bootstrap.servers': testContext.cradle.kafkaConfig.brokers.join(','),
+    })
+
+    // Topics can be created from producers, but as we will first connect a consumer, we need to create the topic first
+    const admin = kafka.admin()
+    await admin.connect()
+    await admin.createTopics({
+      topics: [{ topic }],
+    })
+
+    const messages: string[] = []
+
+    const consumer = kafka.consumer({ 'group.id': groupId })
+    await consumer.connect()
+    await consumer.subscribe({ topic })
+
+    await consumer.run({
+      eachMessage: ({ message }) => {
+        const messageString = message.value?.toString()
+        if (messageString) messages.push(messageString)
+        return Promise.resolve()
+      },
+    })
+    // Wait for the consumer to be assigned partitions
+    await waitAndRetry(() => consumer.assignment().length > 0, 100, 10)
+
+    // When
+    const producer = kafka.producer()
+    await producer.connect()
+    await producer.send({
+      topic,
+      messages: [{ value: messageValue }],
+    })
+
+    // Then
+    await waitAndRetry(() => messages.length > 0)
+
+    // Cleaning up before checks to avoid stale state
+    await consumer.disconnect()
+    await producer.disconnect()
+    await admin.disconnect()
+
+    expect(messages).toHaveLength(1)
+    expect(messages[0]).toEqual(messageValue)
+  })
+})

packages/kafka/lib/test.spec.ts renamed to packages/kafka/lib/test-platformatic.spec.ts

@@ -3,16 +3,14 @@ import { waitAndRetry } from '@lokalise/universal-ts-utils/node'
 import {
   Consumer,
   type Message,
+  ProduceAcks,
   Producer,
   stringDeserializers,
-  stringSerializers,
 } from '@platformatic/kafka'
-import { ProduceAcks } from '@platformatic/kafka'
-import { afterAll } from 'vitest'
+import { stringSerializers } from '@platformatic/kafka'
 import { type TestContext, registerDependencies } from '../test/testContext.ts'
 
-// TODO: to be removed once we have proper tests
-describe('Test', () => {
+describe('Test platformatic-kafka', () => {
   let testContext: TestContext
 
   beforeAll(async () => {
@@ -23,7 +21,7 @@ describe('Test', () => {
     await testContext.dispose()
   })
 
-  it('should send and receive a message', { timeout: 10000 }, async () => {
+  it('should send and receive a message', async () => {
     // Given
     const clientId = randomUUID()
     // Use a fresh, unique topic per run to avoid stale state
@@ -62,12 +60,13 @@ describe('Test', () => {
     })
 
     // Then
-    await waitAndRetry(() => receivedMessages.length > 0, 10, 800)
-    expect(receivedMessages).toHaveLength(1)
-    expect(receivedMessages[0]?.value?.toString()).toBe(messageValue)
+    await waitAndRetry(() => receivedMessages.length > 0)
 
     // Cleanup
     producer.close()
     consumer.close()
+
+    expect(receivedMessages).toHaveLength(1)
+    expect(receivedMessages[0]?.value?.toString()).toBe(messageValue)
   })
 })

packages/kafka/package.json

@@ -1,7 +1,9 @@
 {
     "name": "@message-queue-toolkit/kafka",
     "version": "0.0.0",
-    "engines": { "node": ">= 22.14.0" },
+    "engines": {
+        "node": ">= 22.14.0"
+    },
     "private": false,
     "license": "MIT",
     "homepage": "https://github.com/kibertoad/message-queue-toolkit",
@@ -45,9 +47,10 @@
         "prepublishOnly": "npm run lint && npm run build"
     },
     "dependencies": {
+        "@confluentinc/kafka-javascript": "^1.3.0",
         "@lokalise/node-core": "^14.0.1",
         "@lokalise/universal-ts-utils": "^4.4.1",
-        "@platformatic/kafka": "^1.1.0",
+        "@platformatic/kafka": "^1.2.0",
         "zod": "^3.25.7"
     },
     "peerDependencies": {

packages/kafka/vitest.config.ts

@@ -6,7 +6,6 @@ export default defineConfig({
     globals: true,
     watch: false,
     restoreMocks: true,
-    pool: 'threads',
     coverage: {
       provider: 'v8',
       include: ['lib/**/*.ts'],