From bb522d4e60708aa8fc5db57d0d6824c546b4342a Mon Sep 17 00:00:00 2001 From: Andrey Sobolev Date: Fri, 10 Oct 2025 21:09:57 +0700 Subject: [PATCH] Fix Kafka close of admin --- .../kafka/main_2025-10-10-14-10.json | 10 +++ .../server-core/main_2025-10-10-14-10.json | 10 +++ packages/core/src/queue/types.ts | 2 + packages/kafka/src/__test__/queue.spec.ts | 18 ++++- packages/kafka/src/index.ts | 76 +++++++++++++------ 5 files changed, 88 insertions(+), 28 deletions(-) create mode 100644 common/changes/@hcengineering/kafka/main_2025-10-10-14-10.json create mode 100644 common/changes/@hcengineering/server-core/main_2025-10-10-14-10.json diff --git a/common/changes/@hcengineering/kafka/main_2025-10-10-14-10.json b/common/changes/@hcengineering/kafka/main_2025-10-10-14-10.json new file mode 100644 index 0000000000..fcb16b3805 --- /dev/null +++ b/common/changes/@hcengineering/kafka/main_2025-10-10-14-10.json @@ -0,0 +1,10 @@ +{ + "changes": [ + { + "packageName": "@hcengineering/kafka", + "comment": "Fix kafka close of admin() on create topics", + "type": "patch" + } + ], + "packageName": "@hcengineering/kafka" +} \ No newline at end of file diff --git a/common/changes/@hcengineering/server-core/main_2025-10-10-14-10.json b/common/changes/@hcengineering/server-core/main_2025-10-10-14-10.json new file mode 100644 index 0000000000..9542afd28e --- /dev/null +++ b/common/changes/@hcengineering/server-core/main_2025-10-10-14-10.json @@ -0,0 +1,10 @@ +{ + "changes": [ + { + "packageName": "@hcengineering/server-core", + "comment": "Fix TxOrderingMiddleware", + "type": "patch" + } + ], + "packageName": "@hcengineering/server-core" +} \ No newline at end of file diff --git a/packages/core/src/queue/types.ts b/packages/core/src/queue/types.ts index 618e5511bc..b59b58b32c 100644 --- a/packages/core/src/queue/types.ts +++ b/packages/core/src/queue/types.ts @@ -51,6 +51,8 @@ export interface PlatformQueue { onMessage: (ctx: MeasureContext, msg: ConsumerMessage, queue: ConsumerControl) => Promise, options?: { fromBegining?: boolean + retryDelay?: number // Initial retry delay in milliseconds (default 1000) + maxRetryDelay?: number // Maximum retry delay in seconds (default 10) } ) => ConsumerHandle diff --git a/packages/kafka/src/__test__/queue.spec.ts b/packages/kafka/src/__test__/queue.spec.ts index b9223ac767..ec3840debf 100644 --- a/packages/kafka/src/__test__/queue.spec.ts +++ b/packages/kafka/src/__test__/queue.spec.ts @@ -21,12 +21,18 @@ describe('queue', () => { clearTimeout(to) resolve() } - }) + }, { retryDelay: 100, maxRetryDelay: 5 }) }) const producer = queue.getProducer(testCtx, 'qtest') - for (let i = 0; i < docsCount; i++) { - await producer.send(testCtx, genId as any as WorkspaceUuid, ['msg' + i]) + // Send messages in batches for better performance + const batchSize = 10 + for (let i = 0; i < docsCount; i += batchSize) { + const batch: string[] = [] + for (let j = i; j < Math.min(i + batchSize, docsCount); j++) { + batch.push('msg' + j) + } + await producer.send(testCtx, genId as any as WorkspaceUuid, batch) } await p1 @@ -35,6 +41,8 @@ describe('queue', () => { } finally { await queue.shutdown() await queue.deleteTopics(['test']) + // Give Kafka time to cleanup connections + await new Promise(resolve => setTimeout(resolve, 100)) } }) @@ -51,7 +59,7 @@ describe('queue', () => { throw new Error('Processing Error') } resolve() - }) + }, { retryDelay: 50, maxRetryDelay: 3 }) // Fast retry for tests }) const producer = queue.getProducer(testCtx, 'test') @@ -61,6 +69,8 @@ describe('queue', () => { } finally { await queue.shutdown() await queue.deleteTopics(['test']) + // Give Kafka time to cleanup connections + await new Promise(resolve => setTimeout(resolve, 100)) } }) }) diff --git a/packages/kafka/src/index.ts b/packages/kafka/src/index.ts index 302b23abff..76475f10b7 100644 --- a/packages/kafka/src/index.ts +++ b/packages/kafka/src/index.ts @@ -101,6 +101,8 @@ class PlatformQueueImpl implements PlatformQueue { onMessage: (ctx: MeasureContext, msg: ConsumerMessage, queue: ConsumerControl) => Promise, options?: { fromBegining?: boolean + retryDelay?: number // Initial retry delay in milliseconds (default 1000) + maxRetryDelay?: number // Maximum retry delay in seconds (default 10) } ): ConsumerHandle { const result = new PlatformQueueConsumerImpl(ctx, this.kafka, this.config, topic, groupId, onMessage, options) @@ -111,36 +113,52 @@ class PlatformQueueImpl implements PlatformQueue { async checkCreateTopic (topic: QueueTopic | string, topics: Set, numPartitions?: number): Promise { const kTopic = getKafkaTopicId(topic, this.config) if (!topics.has(kTopic)) { + const admin = this.kafka.admin() try { - await this.kafka.admin().createTopics({ topics: [{ topic: kTopic, numPartitions: numPartitions ?? 1 }] }) + await admin.connect() + await admin.createTopics({ topics: [{ topic: kTopic, numPartitions: numPartitions ?? 1 }] }) } catch (err: any) { console.error('Failed to create topic', kTopic, err) + } finally { + await admin.disconnect() } } } async createTopic (topics: string | string[], partitions: number): Promise { - const existing = new Set(await this.kafka.admin({}).listTopics()) - topics = Array.isArray(topics) ? topics : [topics] - for (const topic of topics) { - await this.checkCreateTopic(topic, existing, partitions) + const admin = this.kafka.admin() + try { + await admin.connect() + const existing = new Set(await admin.listTopics()) + topics = Array.isArray(topics) ? topics : [topics] + for (const topic of topics) { + await this.checkCreateTopic(topic, existing, partitions) + } + } finally { + await admin.disconnect() } } async createTopics (tx: number): Promise { - const topics = new Set(await this.kafka.admin({}).listTopics()) - await this.checkCreateTopic(QueueTopic.Tx, topics, tx) - await this.checkCreateTopic(QueueTopic.Fulltext, topics, 1) - await this.checkCreateTopic(QueueTopic.Workspace, topics, 1) - await this.checkCreateTopic(QueueTopic.Users, topics, 1) - await this.checkCreateTopic(QueueTopic.Process, topics, 1) + const admin = this.kafka.admin() + try { + await admin.connect() + const topics = new Set(await admin.listTopics()) + await this.checkCreateTopic(QueueTopic.Tx, topics, tx) + await this.checkCreateTopic(QueueTopic.Fulltext, topics, 1) + await this.checkCreateTopic(QueueTopic.Workspace, topics, 1) + await this.checkCreateTopic(QueueTopic.Users, topics, 1) + await this.checkCreateTopic(QueueTopic.Process, topics, 1) + } finally { + await admin.disconnect() + } } - async checkDeleteTopic (topic: QueueTopic | string, topics: Set): Promise { + async checkDeleteTopic (admin: any, topic: QueueTopic | string, topics: Set): Promise { const kTopic = getKafkaTopicId(topic, this.config) if (topics.has(kTopic)) { try { - await this.kafka.admin().deleteTopics({ topics: [kTopic] }) + await admin.deleteTopics({ topics: [kTopic] }) } catch (err: any) { console.error('Failed to delete topic', kTopic, err) } @@ -148,16 +166,22 @@ class PlatformQueueImpl implements PlatformQueue { } async deleteTopics (topics?: (QueueTopic | string)[]): Promise { - const existing = new Set(await this.kafka.admin({}).listTopics()) - if (topics !== undefined) { - for (const t of topics) { - await this.checkDeleteTopic(t, existing) + const admin = this.kafka.admin() + try { + await admin.connect() + const existing = new Set(await admin.listTopics()) + if (topics !== undefined) { + for (const t of topics) { + await this.checkDeleteTopic(admin, t, existing) + } + } else { + await this.checkDeleteTopic(admin, QueueTopic.Tx, existing) + await this.checkDeleteTopic(admin, QueueTopic.Fulltext, existing) + await this.checkDeleteTopic(admin, QueueTopic.Workspace, existing) + await this.checkDeleteTopic(admin, QueueTopic.Users, existing) } - } else { - await this.checkDeleteTopic(QueueTopic.Tx, existing) - await this.checkDeleteTopic(QueueTopic.Fulltext, existing) - await this.checkDeleteTopic(QueueTopic.Workspace, existing) - await this.checkDeleteTopic(QueueTopic.Users, existing) + } finally { + await admin.disconnect() } } } @@ -230,6 +254,8 @@ class PlatformQueueConsumerImpl implements ConsumerHandle { ) => Promise, private readonly options?: { fromBegining?: boolean + retryDelay?: number // Initial retry delay in milliseconds (default 1000) + maxRetryDelay?: number // Maximum retry delay in seconds (default 10) } ) { this.cc = this.kafka.consumer({ @@ -253,6 +279,8 @@ class PlatformQueueConsumerImpl implements ConsumerHandle { const meta = JSON.parse(message.headers?.meta?.toString() ?? '{}') const workspace = (message.headers?.workspace?.toString() ?? msgKey) as WorkspaceUuid + const retryDelay = this.options?.retryDelay ?? 1000 + const maxRetryDelay = this.options?.maxRetryDelay ?? 10 let to = 1 while (true) { try { @@ -269,8 +297,8 @@ class PlatformQueueConsumerImpl implements ConsumerHandle { } catch (err: any) { this.ctx.error('failed to process message', { err, msgKey, msgData, workspace }) await heartbeat() - await new Promise((resolve) => setTimeout(resolve, to * 1000)) - if (to < 10) { + await new Promise((resolve) => setTimeout(resolve, to * retryDelay)) + if (to < maxRetryDelay) { to++ } }