From 30b21277312f844eaa9fd46eead8db53a85ab976 Mon Sep 17 00:00:00 2001 From: Thanasis G <96360514+gthanasis@users.noreply.github.com> Date: Fri, 19 Jun 2026 15:13:58 +0300 Subject: [PATCH] fix(Kafka Trigger Node): Surface consumer errors instead of waiting indefinitely (#32642) Co-authored-by: Claude Opus 4.8 (1M context) --- .../nodes/Kafka/KafkaTrigger.node.ts | 12 ++- .../Kafka/test/KafkaTrigger.node.test.ts | 91 ++++++++++++++++++- packages/nodes-base/nodes/Kafka/utils.ts | 45 ++++++++- .../nodes-base/test/nodes/TriggerHelpers.ts | 3 + 4 files changed, 146 insertions(+), 5 deletions(-) diff --git a/packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts b/packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts index 9959f049785..7e2ab081fff 100644 --- a/packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts +++ b/packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts @@ -24,6 +24,8 @@ import { configureDataEmitter, getAutoCommitSettings, runWithHeartbeat, + toUserFacingConsumerError, + type ConsumerErrorHandler, } from './utils'; export class KafkaTrigger implements INodeType { @@ -485,9 +487,17 @@ export class KafkaTrigger implements INodeType { } }; - const listeners = connectEventListeners(consumer, this.logger); + let closeGotCalled = false; + const handleConsumerError: ConsumerErrorHandler = (error) => { + // Don't surface errors that are a side effect of our own teardown. + if (!closeGotCalled) { + this.emitError(toUserFacingConsumerError(this.getNode(), error)); + } + }; + const listeners = connectEventListeners(consumer, this.logger, handleConsumerError); const closeFunction = async () => { + closeGotCalled = true; try { disconnectEventListeners(listeners); await consumer.stop(); diff --git a/packages/nodes-base/nodes/Kafka/test/KafkaTrigger.node.test.ts b/packages/nodes-base/nodes/Kafka/test/KafkaTrigger.node.test.ts index 420f902f547..f2e25d31e51 100644 --- a/packages/nodes-base/nodes/Kafka/test/KafkaTrigger.node.test.ts +++ b/packages/nodes-base/nodes/Kafka/test/KafkaTrigger.node.test.ts @@ -39,8 +39,12 @@ describe('KafkaTrigger Node', () => { let mockConsumerCreate: Mock; let mockRegistryDecode: Mock; let publishMessage: (message: Partial) => Promise; + // Records the handlers registered via consumer.on(event, handler) so tests can + // fire lifecycle events (e.g. a consumer crash) the way kafkajs would at runtime. + let consumerEventHandlers: Record unknown>; beforeEach(() => { + consumerEventHandlers = {}; const mockEachMessageHolder = { handler: vi.fn(async () => {}) as Mocked, }; @@ -64,7 +68,10 @@ describe('KafkaTrigger Node', () => { subscribe: mockConsumerSubscribe, run: mockConsumerRun, disconnect: mockConsumerDisconnect, - on: vi.fn(() => vi.fn()), + on: vi.fn((event: string, handler: (event: unknown) => unknown) => { + consumerEventHandlers[event] = handler; + return vi.fn(); + }), events: { CONNECT: 'consumer.connect', GROUP_JOIN: 'consumer.group_join', @@ -218,6 +225,88 @@ describe('KafkaTrigger Node', () => { expect(mockConsumerDisconnect).toHaveBeenCalled(); }); + // A Kafka Trigger that loses its consumer after start-up (e.g. a compressed + // batch it cannot decode throws a non-retriable error, or a group-ACL denial) + // must surface an error to the execution instead of hanging on "Waiting..." + // forever. Non-retriable crashes are routed to emitError; retriable ones are + // left to the client's auto-restart. + describe('surfaces async consumer failures instead of hanging', () => { + const fireCrash = async ( + restart: boolean, + message = 'KafkaJSNotImplemented: LZ4 compression not implemented', + ) => { + const crashHandler = consumerEventHandlers['consumer.crash']; + expect(crashHandler).toBeDefined(); + await crashHandler({ + id: 0, + type: 'consumer.crash', + timestamp: 0, + payload: { + error: new Error(message), + groupId: 'test-group', + restart, + }, + }); + }; + + const startTrigger = async () => + await testTriggerNode(KafkaTrigger, { + mode: 'trigger', + node: { + typeVersion: 1.3, + parameters: { + topic: 'test-topic', + groupId: 'test-group', + useSchemaRegistry: false, + options: { fromBeginning: true }, + }, + }, + credential: { + brokers: 'localhost:9092', + clientId: 'n8n-kafka', + ssl: true, + authentication: false, + }, + }); + + it('surfaces an unsupported-compression crash with an actionable error', async () => { + const { emitError } = await startTrigger(); + + await fireCrash(false); + + expect(emitError).toHaveBeenCalledTimes(1); + const surfaced = emitError.mock.calls[0][0]; + expect(surfaced).toBeInstanceOf(NodeOperationError); + expect(surfaced.message).toMatch(/unsupported compression codec/i); + }); + + it('surfaces other non-retriable crashes with the original error', async () => { + const { emitError } = await startTrigger(); + + await fireCrash(false, 'Broker: Group authorization failed'); + + expect(emitError).toHaveBeenCalledTimes(1); + expect(emitError.mock.calls[0][0].message).toBe('Broker: Group authorization failed'); + }); + + it('does not surface a retriable crash (kafkajs auto-restarts)', async () => { + const { emitError } = await startTrigger(); + + await fireCrash(true); + + expect(emitError).not.toHaveBeenCalled(); + }); + + it('does not surface a crash that happens during teardown', async () => { + const { emitError, close } = await startTrigger(); + + await close(); + await fireCrash(false); + + expect(emitError).not.toHaveBeenCalled(); + }); + }); + it('should handle authentication when credentials are provided', async () => { await testTriggerNode(KafkaTrigger, { mode: 'trigger', diff --git a/packages/nodes-base/nodes/Kafka/utils.ts b/packages/nodes-base/nodes/Kafka/utils.ts index 47b3d5e5cda..0732e19c2dc 100644 --- a/packages/nodes-base/nodes/Kafka/utils.ts +++ b/packages/nodes-base/nodes/Kafka/utils.ts @@ -14,6 +14,7 @@ import type { IDataObject, IRun, IBinaryKeyData, + INode, INodeExecutionData, FunctionsBase, RequestHelperFunctions, @@ -229,13 +230,43 @@ export function configureMessageParser( }; } +/** + * Maps a fatal consumer error to a user-facing error. Known cases (such as a + * topic compressed with a codec the client cannot decode) get an actionable + * message that points the user at a fix; anything else is surfaced unchanged so + * the original failure is still shown instead of an indefinite wait. + * @param node - The node raising the error + * @param error - The fatal error from the consumer + */ +export function toUserFacingConsumerError(node: INode, error: Error): Error { + if (/compression not implemented/i.test(error.message)) { + return new NodeOperationError(node, 'Kafka topic uses an unsupported compression codec', { + description: + 'This topic contains messages compressed with LZ4, Snappy, or ZSTD, which the Kafka Trigger cannot decode (only GZIP and uncompressed messages are supported). Set the producer to use gzip or no compression to consume this topic.', + }); + } + + return error; +} + +/** + * Handler invoked with a fatal (non-retriable) consumer error so the caller can + * surface it to the execution instead of leaving the trigger waiting. + */ +export type ConsumerErrorHandler = (error: Error) => void; + /** * Attaches event listeners to the Kafka consumer for monitoring and logging * @param consumer - The Kafka consumer instance * @param logger - Logger instance for event logging + * @param onFatalCrash - Optional handler called when the consumer crashes non-retriably * @returns Array of listener removal functions */ -export function connectEventListeners(consumer: Consumer, logger: Logger) { +export function connectEventListeners( + consumer: Consumer, + logger: Logger, + onFatalCrash?: ConsumerErrorHandler, +) { const onConnected = consumer.on(consumer.events.CONNECT, () => { logger.debug('Kafka consumer connected'); }); @@ -263,8 +294,16 @@ export function connectEventListeners(consumer: Consumer, logger: Logger) { const onRebalancing = consumer.on(consumer.events.REBALANCING, (payload) => { logger.debug('Consumer is rebalancing', { payload }); }); - const onCrash = consumer.on(consumer.events.CRASH, async (error) => { - logger.error('Consumer has crashed', { error }); + const onCrash = consumer.on(consumer.events.CRASH, (event) => { + const { error, restart } = event.payload; + logger.error('Consumer has crashed', { error, restart }); + // kafkajs auto-restarts retriable crashes (restart === true). A non-retriable + // crash (e.g. an undecodable compressed batch, or a group authorization + // failure) leaves the consumer dead; without surfacing it, the trigger just + // keeps waiting forever. Route it to the caller so the execution can fail. + if (!restart) { + onFatalCrash?.(ensureError(error)); + } }); return [ diff --git a/packages/nodes-base/test/nodes/TriggerHelpers.ts b/packages/nodes-base/test/nodes/TriggerHelpers.ts index 6d26992fff1..ec20aa56423 100644 --- a/packages/nodes-base/test/nodes/TriggerHelpers.ts +++ b/packages/nodes-base/test/nodes/TriggerHelpers.ts @@ -75,6 +75,7 @@ export async function testTriggerNode( ) { const trigger = 'description' in Trigger ? Trigger : new Trigger(); const emit: MockedFunction = vi.fn(); + const emitError: MockedFunction = vi.fn(); const timezone = options.timezone ?? 'Europe/Berlin'; const version = trigger.description.version; @@ -123,6 +124,7 @@ export async function testTriggerNode( const triggerFunctions = mock({ helpers, emit, + emitError, logger: triggerLogger, getTimezone: () => timezone, getNode: () => node, @@ -145,6 +147,7 @@ export async function testTriggerNode( close: vi.fn(response?.closeFunction), manualTriggerFunction: options.mode === 'manual' ? response?.manualTriggerFunction : undefined, emit, + emitError, logger: triggerLogger, }; }