Bump @platformatic/kafka version (#484)

* Bump platformatic/kafka version

* Fix topics deletion issue

Author: kjamrog

packages/kafka/package.json

@@ -2,7 +2,7 @@
     "name": "@message-queue-toolkit/kafka",
     "version": "0.11.4",
     "engines": {
-        "node": ">= 22.14.0"
+        "node": ">= 22.22.0"
     },
     "private": false,
     "license": "MIT",
@@ -53,7 +53,7 @@
     "dependencies": {
         "@lokalise/node-core": "^14.8.1",
         "@lokalise/universal-ts-utils": "^4.10.0",
-        "@platformatic/kafka": "^2.2.3"
+        "@platformatic/kafka": "^2.3.1"
     },
     "peerDependencies": {
         "@message-queue-toolkit/core": ">=26.1.1",

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

@@ -14,7 +14,7 @@ import {
   type PermissionAdded,
   TOPICS,
 } from '../utils/permissionSchemas.ts'
-import { createTestContext, type TestContext } from '../utils/testContext.ts'
+import { createTestContext, deleteExistingTopics, type TestContext } from '../utils/testContext.ts'
 import { PermissionBatchConsumer } from './PermissionBatchConsumer.ts'
 
 describe('PermissionBatchConsumer', () => {
@@ -35,13 +35,7 @@ describe('PermissionBatchConsumer', () => {
 
   describe('init - close', () => {
     beforeEach(async () => {
-      try {
-        await testContext.cradle.kafkaAdmin.deleteTopics({
-          topics: TOPICS,
-        })
-      } catch (_) {
-        // Ignore errors if the topic does not exist
-      }
+      await deleteExistingTopics(testContext.cradle.kafkaAdmin, TOPICS)
     })
 
     it('should thrown an error if topics is empty', async () => {

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

@@ -11,7 +11,7 @@ import {
   type PermissionAdded,
   TOPICS,
 } from '../utils/permissionSchemas.ts'
-import { createTestContext, type TestContext } from '../utils/testContext.ts'
+import { createTestContext, deleteExistingTopics, type TestContext } from '../utils/testContext.ts'
 import { PermissionConsumer } from './PermissionConsumer.ts'
 
 describe('PermissionConsumer', () => {
@@ -32,13 +32,7 @@ describe('PermissionConsumer', () => {
 
   describe('init - close', () => {
     beforeEach(async () => {
-      try {
-        await testContext.cradle.kafkaAdmin.deleteTopics({
-          topics: TOPICS,
-        })
-      } catch (_) {
-        // Ignore errors if the topic does not exist
-      }
+      await deleteExistingTopics(testContext.cradle.kafkaAdmin, TOPICS)
     })
 
     it('should thrown an error if topics is empty', async () => {

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

@@ -7,7 +7,7 @@ import {
   type PermissionRemoved,
   TOPICS,
 } from '../utils/permissionSchemas.ts'
-import { createTestContext, type TestContext } from '../utils/testContext.ts'
+import { createTestContext, deleteExistingTopics, type TestContext } from '../utils/testContext.ts'
 import { PermissionPublisher } from './PermissionPublisher.ts'
 
 describe('PermissionPublisher', () => {
@@ -28,13 +28,7 @@ describe('PermissionPublisher', () => {
 
   describe('init - close', () => {
     beforeEach(async () => {
-      try {
-        await testContext.cradle.kafkaAdmin.deleteTopics({
-          topics: TOPICS,
-        })
-      } catch (_) {
-        // Ignore errors if the topic does not exist
-      }
+      await deleteExistingTopics(testContext.cradle.kafkaAdmin, TOPICS)
     })
 
     it('should thrown an error if topics is empty', () => {

packages/kafka/test/utils/testContext.ts

@@ -55,6 +55,23 @@ export const getKafkaConfig = (): KafkaConfig => ({
   clientId: randomUUID(),
 })
 
+/**
+ * Deletes only the topics that currently exist.
+ *
+ * Workaround for a regression in @platformatic/kafka >= 2.3.0 where Admin#deleteTopics
+ * leaves a stale entry in its internal deduplication cache when the operation errors
+ * (e.g. deleting a non-existent topic). Reusing the same Admin instance afterwards causes
+ * subsequent deleteTopics calls for the same topics to hang. Filtering to existing topics
+ * avoids triggering that error path.
+ */
+export const deleteExistingTopics = async (kafkaAdmin: Admin, topics: string[]): Promise<void> => {
+  const existingTopics = await kafkaAdmin.listTopics()
+  const topicsToDelete = topics.filter((topic) => existingTopics.includes(topic))
+  if (topicsToDelete.length === 0) return
+
+  await kafkaAdmin.deleteTopics({ topics: topicsToDelete })
+}
+
 const resolveDIConfig = (awilixManager: AwilixManager): DiConfig => ({
   awilixManager: asFunction(() => awilixManager, SINGLETON_CONFIG),
   kafkaConfig: asFunction(getKafkaConfig, SINGLETON_CONFIG),