fix(Kafka Trigger Node): Fail v2 activation when the consumer group cannot be joined (#36046)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

packages/nodes-base/nodes/Kafka/test/mocks/confluent-kafka.ts

@@ -28,6 +28,9 @@ export interface FakeConsumer {
 	// No `stop`: the real consumer's `stop()` calls notImplemented() and throws, so
 	// a fake that resolves would let code depend on something that cannot work.
 	disconnect: Mock;
+	/** Joined by default, so the startup join wait exits on its first check.
+	 * Override via {@link setFakeConsumerAssignment}. */
+	assignment: Mock;
 	/** Feeds a batch through the handler `run()` registered. */
 	deliverBatch: (batch: FakeBatch) => Promise<void>;
 	/** Spies on the `EachBatchPayload` callbacks handed to that handler. */
@@ -42,6 +45,20 @@ export interface FakeConsumer {
 const consumers: FakeConsumer[] = [];
 const clientConfigs: KafkaJS.CommonConstructorConfig[] = [];
 
+/** Joined by default, so the startup join wait costs the suites nothing. */
+const joinedAssignment = () => [{ topic: 'test-topic', partition: 0 }];
+let nextAssignment: () => Array<{ topic: string; partition: number }> = joinedAssignment;
+
+/**
+ * Makes every fake consumer report this assignment, e.g. `() => []` for one
+ * that never joins its group. Reset by {@link resetConfluentKafkaRecordings}.
+ */
+export function setFakeConsumerAssignment(
+	assignment: () => Array<{ topic: string; partition: number }>,
+): void {
+	nextAssignment = assignment;
+}
+
 function createFakeConsumer(config: KafkaJS.ConsumerConstructorConfig): FakeConsumer {
 	let eachBatch: EachBatchHandler | undefined;
 
@@ -61,6 +78,7 @@ function createFakeConsumer(config: KafkaJS.ConsumerConstructorConfig): FakeCons
 			eachBatch = runConfig?.eachBatch;
 		}),
 		disconnect: vi.fn(async () => {}),
+		assignment: vi.fn(() => nextAssignment()),
 		payloadSpies,
 		deliverBatch: async ({
 			messages,
@@ -156,4 +174,5 @@ export function getFakeClientConfigs(): KafkaJS.CommonConstructorConfig[] {
 export function resetConfluentKafkaRecordings(): void {
 	consumers.length = 0;
 	clientConfigs.length = 0;
+	nextAssignment = joinedAssignment;
 }

packages/nodes-base/nodes/Kafka/test/v2/KafkaTriggerV2.node.test.ts

@@ -18,6 +18,7 @@ import {
 	confluentKafkaModuleMock,
 	getFakeConsumers,
 	resetConfluentKafkaRecordings,
+	setFakeConsumerAssignment,
 	type FakeConsumer,
 } from '../mocks/confluent-kafka';
 
@@ -33,8 +34,8 @@ vi.mock('../../v2/consumer', async (importOriginal) => {
 	return {
 		...actual,
 		consumeTopic: vi.fn(async (...args: Parameters<typeof actual.consumeTopic>) => {
-			consumeTopicSpy(...args);
-			return await actual.consumeTopic(...args);
+			consumeTopicSpy.apply(undefined, args);
+			return await actual.consumeTopic.apply(actual, args);
 		}),
 	};
 });
@@ -637,6 +638,81 @@ describe('KafkaTriggerV2 Node', () => {
 
 			expect(emitError).not.toHaveBeenCalled();
 		});
+
+		describe('while startup is still waiting on the group join', () => {
+			// A fatal here must fail activation instead of reaching emitError, which
+			// would flap deactivate-reactivate with no backoff (ENT-340).
+			beforeEach(() => {
+				vi.useFakeTimers();
+				setFakeConsumerAssignment(() => []);
+			});
+
+			afterEach(() => {
+				vi.useRealTimers();
+			});
+
+			it('fails activation instead of reporting a successful start', async () => {
+				const starting = startTrigger('v2-unjoinable');
+				await vi.advanceTimersByTimeAsync(0);
+				const consumer = await lastFakeConsumer();
+
+				libraryLogger(consumer).error('Broker: Group authorization failed');
+
+				await expect(starting).rejects.toThrow(/authorization failed/i);
+				expect(consumer.disconnect).toHaveBeenCalledTimes(1);
+			});
+
+			it('keeps the manual-run ACL explanation when the denial fails the join wait', async () => {
+				// The rewritten ACL hint must survive the startup-failure path; the raw
+				// broker message names neither the throwaway group nor the fix.
+				const started = await testTriggerNode(new KafkaTriggerV2(baseDescription), {
+					mode: 'manual',
+					node: {
+						parameters: {
+							topic: 'test-topic',
+							groupId: 'orders-consumer',
+							useSchemaRegistry: false,
+						},
+					},
+					credential,
+				});
+
+				const starting = started.manualTriggerFunction?.();
+				await vi.advanceTimersByTimeAsync(0);
+				const consumer = await lastFakeConsumer();
+
+				libraryLogger(consumer).error('Broker: Group authorization failed');
+
+				await expect(starting).rejects.toMatchObject({
+					message: expect.stringContaining(
+						'Kafka refused the consumer group used for a test run',
+					) as string,
+					description: expect.stringContaining('orders-consumer-n8n-manual-') as string,
+				});
+				expect(started.emitError).not.toHaveBeenCalled();
+			});
+
+			it('closes the consumer when a manual run is cancelled during the join wait', async () => {
+				const started = await testTriggerNode(new KafkaTriggerV2(baseDescription), {
+					mode: 'manual',
+					node: {
+						parameters: { topic: 'test-topic', groupId: 'v2-join-wait', useSchemaRegistry: false },
+					},
+					credential,
+				});
+
+				// Cancel while the start is held open by the join wait.
+				const starting = started.manualTriggerFunction?.();
+				await vi.advanceTimersByTimeAsync(0);
+				const closing = started.close?.();
+				await vi.advanceTimersByTimeAsync(3000);
+				await Promise.all([starting, closing]);
+
+				const consumer = await lastFakeConsumer();
+				expect(consumer.disconnect).toHaveBeenCalled();
+				expect(started.emitError).not.toHaveBeenCalled();
+			});
+		});
 	});
 
 	describe('message shape options', () => {

packages/nodes-base/nodes/Kafka/test/v2/consumer/ConsumeTopic.test.ts

@@ -118,6 +118,90 @@ describe('consumeTopic', () => {
 		});
 	});
 
+	describe('group join wait', () => {
+		// connect/subscribe/run resolve before the join settles, so startup holds
+		// until the outcome is known (ENT-340).
+		beforeEach(() => {
+			vi.useFakeTimers();
+		});
+
+		afterEach(() => {
+			vi.useRealTimers();
+		});
+
+		it('resolves as soon as the group assigns partitions', async () => {
+			const consumer = await newConsumer();
+			// Not joined on the first check, joined on the next poll.
+			consumer.assignment.mockImplementationOnce(() => []);
+
+			const startup = consumeTopic(consumer as never, {
+				topic: 'orders',
+				parseMessage,
+				emit,
+				logger,
+			});
+
+			// One poll interval, nowhere near the full grace period.
+			await vi.advanceTimersByTimeAsync(250);
+			await expect(startup).resolves.toBeDefined();
+		});
+
+		it('fails startup and disconnects when a fatal error arrives while unjoined', async () => {
+			const consumer = await newConsumer();
+			consumer.assignment.mockImplementation(() => []);
+			let failStartup!: (error: Error) => void;
+			const startupFailure = new Promise<never>((_, reject) => (failStartup = reject));
+
+			const startup = consumeTopic(consumer as never, {
+				topic: 'orders',
+				parseMessage,
+				emit,
+				logger,
+				startupFailure,
+			});
+			failStartup(new Error('Broker: Group authorization failed'));
+
+			await expect(startup).rejects.toThrow('Group authorization failed');
+			expect(consumer.disconnect).toHaveBeenCalledTimes(1);
+		});
+
+		it('proceeds after the grace period when the group has assigned nothing', async () => {
+			// Zero partitions can be legitimate, so only an observed fatal error may
+			// fail startup, never the clock.
+			const consumer = await newConsumer();
+			consumer.assignment.mockImplementation(() => []);
+
+			const startup = consumeTopic(consumer as never, {
+				topic: 'orders',
+				parseMessage,
+				emit,
+				logger,
+			});
+
+			await vi.advanceTimersByTimeAsync(3000);
+			await expect(startup).resolves.toBeDefined();
+			expect(consumer.disconnect).not.toHaveBeenCalled();
+		});
+
+		it('treats an assignment read that throws as not joined, not as a failure', async () => {
+			// The real assignment() throws ERR__STATE unless the consumer is CONNECTED.
+			const consumer = await newConsumer();
+			consumer.assignment.mockImplementation(() => {
+				throw new Error('Assignment can only be called while connected.');
+			});
+
+			const startup = consumeTopic(consumer as never, {
+				topic: 'orders',
+				parseMessage,
+				emit,
+				logger,
+			});
+
+			await vi.advanceTimersByTimeAsync(3000);
+			await expect(startup).resolves.toBeDefined();
+		});
+	});
+
 	describe('chunking', () => {
 		it('emits one execution per message by default', async () => {
 			const { consumer } = await start();
@@ -257,7 +341,7 @@ describe('consumeTopic', () => {
 				await consumer.deliverBatch({ messages: messages('a', 'b') });
 
 				// Every message reaches a workflow, whatever the setting was.
-				expect(emit.mock.calls.flatMap((call) => call[0] as INodeExecutionData[])).toHaveLength(2);
+				expect(emit.mock.calls.flatMap((call) => call[0])).toHaveLength(2);
 				expect(consumer.payloadSpies.resolveOffset).toHaveBeenCalledWith('1');
 			},
 		);

packages/nodes-base/nodes/Kafka/test/v2/kafka.integration.test.ts

@@ -16,6 +16,7 @@
  * manual evidence on ENT-222.
  */
 import { sleep } from '@n8n/utils/sleep';
+import { Kafka, type Consumer } from 'kafkajs';
 import { createServiceStack, type N8NStack } from 'n8n-containers';
 import type {
 	IBinaryData,
@@ -258,6 +259,76 @@ describe('delivery guarantees against a real broker', () => {
 	}, 180_000);
 });
 
+describe('a group the consumer can never join fails startup (ENT-340)', () => {
+	let kafkajsConsumer: Consumer | undefined;
+
+	afterEach(async () => {
+		await kafkajsConsumer?.disconnect().catch(() => {});
+		kafkajsConsumer = undefined;
+	});
+
+	it('rejects with the broker refusal instead of reporting a successful start', async () => {
+		// Staged as the bug was found: kafkajs (v1) and librdkafka (v2) advertise
+		// different partition-assignment strategy names, so whichever joins second
+		// is refused with "Broker: Inconsistent group protocol" forever.
+		const topic = uniqueTopic('join-refused');
+		const groupId = `${topic}-group`;
+		await createTopic(topic);
+
+		// The incumbent: a kafkajs consumer holding the group, as v1 would.
+		kafkajsConsumer = new Kafka({ clientId: 'ent340-v1', brokers: [credentials.brokers] }).consumer(
+			{ groupId },
+		);
+		await kafkajsConsumer.connect();
+		await kafkajsConsumer.subscribe({ topic });
+		await kafkajsConsumer.run({ eachMessage: async () => {} });
+		await withDeadline(
+			(async () => {
+				while (
+					!(
+						await inBroker(
+							`kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group ${groupId} --state`,
+						)
+					).includes('Stable')
+				) {
+					await sleep(500);
+				}
+			})(),
+			30_000,
+			'the kafkajs consumer to hold the group',
+		);
+
+		// The challenger: a v2 consumer wired the way the node wires it.
+		let failStartup!: (error: Error) => void;
+		const startupFailure = new Promise<never>((_, reject) => (failStartup = reject));
+		void startupFailure.catch(() => {});
+
+		const consumer = await createKafkaConsumer(
+			credentials,
+			{ groupId },
+			{ logger, onFatalError: (error) => failStartup(error) },
+		);
+
+		const startup = consumeTopic(consumer, {
+			topic,
+			logger,
+			parseMessage: createMessageParser({}, logger, undefined, prepareBinaryData),
+			emit: async () => ({ mayAdvance: true }),
+			startupFailure,
+		});
+
+		await expect(withDeadline(startup, 30_000, 'startup to settle')).rejects.toThrow(
+			/inconsistent group protocol/i,
+		);
+
+		// The refused consumer must not linger in the group.
+		const members = await inBroker(
+			`kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group ${groupId} --state`,
+		);
+		expect(members).toMatch(/Stable\s+1\s*$/m);
+	}, 120_000);
+});
+
 describe('library logging against a real broker', () => {
 	/** A library logger that records the levels the library asks it to apply. */
 	const recordingLogger = (nodeLogger: Logger, onFatalError?: (error: Error) => void) => {

packages/nodes-base/nodes/Kafka/test/v2/transport/LibraryLogger.test.ts

@@ -40,11 +40,13 @@ describe('createLibraryLogger', () => {
 			'Broker: Topic authorization failed',
 			'Local: Authentication failure',
 			'Broker: SASL Authentication failed',
+			'Broker: Inconsistent group protocol',
+			'Broker: Invalid session timeout',
 		])('reports %s as fatal', (message) => {
 			build().error(message);
 
 			expect(onFatalError).toHaveBeenCalledTimes(1);
-			const reported = onFatalError.mock.calls[0][0] as Error;
+			const reported = onFatalError.mock.calls[0][0];
 			expect(reported).toBeInstanceOf(UserError);
 			expect(reported.message).toBe(message);
 		});

packages/nodes-base/nodes/Kafka/v2/KafkaTriggerV2.node.ts

@@ -12,8 +12,8 @@ import { setSchemaRegistry, type KafkaCredentials } from '../utils';
 import { consumeTopic, createDataEmitter, createMessageParser } from './consumer';
 import type { KafkaConsumerHandle } from './consumer';
 import { versionDescription } from './KafkaTriggerV2Description';
-import { explainManualRunGroupDenial, getSettings } from './TriggerSettings';
 import { createKafkaConsumer } from './transport';
+import { explainManualRunGroupDenial, getSettings } from './TriggerSettings';
 
 export class KafkaTriggerV2 implements INodeType {
 	description: INodeTypeDescription;
@@ -61,6 +61,16 @@ export class KafkaTriggerV2 implements INodeType {
 		};
 
 		const startConsumerOnce = async () => {
+			// Where a fatal consumer error goes depends on when it arrives (ENT-340):
+			// - during startup: reject this gate → activation fails and n8n retries
+			//   with backoff, instead of flapping through emitError every second
+			// - after a successful start: emitError → n8n restarts the dead trigger
+			// - after a failed start: log it; nothing activated, nothing to restart
+			let reportFatal!: (error: Error) => void;
+			const startupFailure = new Promise<never>((_, reject) => (reportFatal = reject));
+			// A rejection that loses the startup race would otherwise be unhandled.
+			void startupFailure.catch(() => {});
+
 			try {
 				const consumer = await createKafkaConsumer(credentials, settings.consumer, {
 					logger: this.logger,
@@ -69,7 +79,7 @@ export class KafkaTriggerV2 implements INodeType {
 					// not failures, so they stay quiet.
 					onFatalError: (error) => {
 						if (closeController.signal.aborted) return;
-						this.emitError(
+						reportFatal(
 							explainManualRunGroupDenial(error, settings.configuredGroupId, settings.isManualRun),
 						);
 					},
@@ -83,8 +93,18 @@ export class KafkaTriggerV2 implements INodeType {
 					batchSize: settings.batchSize,
 					partitionsConsumedConcurrently: settings.partitionsConsumedConcurrently,
 					errorRetryDelay: settings.errorRetryDelay,
+					startupFailure,
+					closeSignal: closeController.signal,
 				});
+				// Startup succeeded. No gap here: a fatal always arrives from a fresh
+				// macrotask, so it cannot land between the await and this re-point.
+				reportFatal = (error) => this.emitError(error);
 			} catch (error) {
+				// Startup failed.
+				reportFatal = (fatal) =>
+					this.logger.error('Kafka consumer reported a fatal error after startup had failed', {
+						error: fatal,
+					});
 				throw new NodeOperationError(this.getNode(), error);
 			}
 		};

packages/nodes-base/nodes/Kafka/v2/TriggerSettings.ts

@@ -95,7 +95,8 @@ function heartbeatWithinSession(
 	sessionTimeout: number,
 	logger?: Logger,
 ): number {
-	// An unusable session timeout is left to librdkafka to reject by name.
+	// An unusable session timeout is left to the broker, which refuses the join;
+	// `transport/LibraryLogger` matches that refusal as non-recoverable.
 	if (!Number.isFinite(sessionTimeout) || sessionTimeout <= 0) return heartbeatInterval;
 
 	const largest = Math.floor(sessionTimeout / HEARTBEATS_PER_SESSION);

packages/nodes-base/nodes/Kafka/v2/consumer/ConsumeTopic.ts

@@ -26,6 +26,18 @@ export interface ConsumeTopicOptions {
 	 * only to failed offset resolution, so the parse path was unpaced.
 	 */
 	errorRetryDelay?: number;
+	/**
+	 * Rejects when the transport reports a non-recoverable error, so a consumer
+	 * that can never join its group fails startup instead of failing silently
+	 * after a nominal start (ENT-340).
+	 */
+	startupFailure?: Promise<never>;
+	/**
+	 * The trigger's own close signal. Folded into the internal one so a cancel
+	 * during startup ends the join wait, instead of the caller waiting out the
+	 * grace period before teardown can begin.
+	 */
+	closeSignal?: AbortSignal;
 }
 
 export interface KafkaConsumerHandle {
@@ -63,6 +75,12 @@ interface BatchContext extends ConsumeSettings {
 /** Bounds teardown so a hung broker request cannot block deactivation, as in v1. */
 const CLOSE_TIMEOUT_MS = 30_000;
 
+/** How often the startup join wait re-checks the consumer's assignment. */
+const JOIN_POLL_INTERVAL_MS = 250;
+
+/** How long startup waits for the group join outcome before proceeding anyway. */
+const JOIN_GRACE_PERIOD_MS = 3000;
+
 /**
  * Messages handed to one execution. v1's Batch Size default, and the reason the
  * library's batch is chunked rather than emitted whole: at 1 each message starts
@@ -101,7 +119,10 @@ export async function consumeTopic(
 	const settings = resolveSettings(options);
 
 	const closeController = new AbortController();
-	const { signal } = closeController;
+
+	const signal = options.closeSignal
+		? AbortSignal.any([closeController.signal, options.closeSignal])
+		: closeController.signal;
 
 	// Turns "we are closing" into something a wait can race against, since you can
 	// race a promise but not a signal. It only ever resolves: the loser of a race
@@ -139,13 +160,20 @@ export async function consumeTopic(
 			eachBatchAutoResolve: false,
 			eachBatch: async (payload) => await processBatch(payload, context),
 		});
+
+		await waitForGroupJoin(consumer, signal, closed, options.startupFailure);
 	} catch (error) {
 		// Nothing else holds this consumer yet, so a failed start must not leave the
 		// broker connection open. `connect()` is inside the try for symmetry rather
 		// than because it leaks: the library marks a failed connect as disconnected
 		// before rejecting, so disconnect() is a no-op on that path today.
 		try {
-			await consumer.disconnect();
+			closeController.abort();
+			await withTimeout(
+				consumer.disconnect(),
+				CLOSE_TIMEOUT_MS,
+				'Kafka consumer did not disconnect in time',
+			);
 		} catch {
 			// The start failure is the useful one; a failing disconnect must not mask it.
 		}
@@ -166,6 +194,39 @@ export async function consumeTopic(
 	};
 }
 
+/**
+ * Holds startup until the consumer joins its group, a non-recoverable error
+ * surfaces, or the grace period elapses. The library resolves connect,
+ * subscribe and run before the join settles, so without this wait a group the
+ * consumer can never join still reported a successful start (ENT-340).
+ *
+ * Expiry is deliberately success: zero partitions can be legitimate (more
+ * members than partitions, a slow rebalance), so only an observed fatal error
+ * may fail startup, never the clock. A close ends the wait the same way, so the
+ * caller reaches its teardown rather than paying out the grace period first.
+ */
+async function waitForGroupJoin(
+	consumer: KafkaJS.Consumer,
+	signal: AbortSignal,
+	closed: Promise<void>,
+	startupFailure?: Promise<never>,
+): Promise<void> {
+	const pollUntilJoined = async () => {
+		const deadline = Date.now() + JOIN_GRACE_PERIOD_MS;
+		while (Date.now() < deadline && !signal.aborted) {
+			try {
+				if (consumer.assignment().length > 0) return;
+			} catch {
+				// assignment() throws ERR__STATE while not connected; means "not joined".
+			}
+			await Promise.race([sleep(JOIN_POLL_INTERVAL_MS), closed]);
+		}
+	};
+	// Race the whole poll, not each pause: a fatal that fired first must win
+	// even if a later poll would see a joined group.
+	await (startupFailure ? Promise.race([pollUntilJoined(), startupFailure]) : pollUntilJoined());
+}
+
 // ---------------------------------------------------------------------------
 // The batch loop
 // ---------------------------------------------------------------------------

packages/nodes-base/nodes/Kafka/v2/transport/LibraryLogger.ts

@@ -17,8 +17,22 @@ import { UserError } from 'n8n-workflow';
  * no error code by the time it reaches a logger (`_consumer.js` builds the entry
  * from `err.message` alone) and exposes no error event, so this is the only hook
  * there is. v1 does the same thing in `toUserFacingConsumerError`.
+ *
+ * Not everything permanent is even visible here: a join loop rejected with
+ * `UNKNOWN_MEMBER_ID` is `ERR_ACTION_IGNORE`d by librdkafka and logs nothing, so
+ * no pattern can catch it.
  */
-const NON_RECOVERABLE = [/authorization failed/i, /authentication fail/i];
+const NON_RECOVERABLE = [
+	/authorization failed/i,
+	/authentication fail/i,
+	// Group members advertise incompatible partition-assignment strategies
+	// (e.g. kafkajs and librdkafka sharing a group); no retry can succeed.
+	/inconsistent group protocol/i,
+	// A Session Timeout inside librdkafka's own range but outside the broker's
+	// [group.min, group.max] window: the broker refuses every JoinGroup, and
+	// librdkafka masks retry off as permanent (`rdkafka_request.c:173`).
+	/invalid session timeout/i,
+];
 
 function isNonRecoverable(message: string): boolean {
 	return NON_RECOVERABLE.some((pattern) => pattern.test(message));