fix(Kafka Trigger Node): Refuse activation when the topic does not exist (#36044)

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

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

@@ -42,7 +42,21 @@ export interface FakeConsumer {
 	};
 }
 
+/** The admin client `assertTopicExists` uses to verify the topic. */
+export interface FakeAdmin {
+	connect: Mock;
+	fetchTopicMetadata: Mock;
+	disconnect: Mock;
+}
+
+/** librdkafka's code for a topic the broker does not know, as the real library exposes it. */
+export const UNKNOWN_TOPIC_OR_PART = 3;
+
+/** librdkafka's code for a topic name the broker rejects outright (also used for patterns). */
+export const TOPIC_EXCEPTION = 17;
+
 const consumers: FakeConsumer[] = [];
+const admins: FakeAdmin[] = [];
 const clientConfigs: KafkaJS.CommonConstructorConfig[] = [];
 
 /** Joined by default, so the startup join wait costs the suites nothing. */
@@ -59,6 +73,67 @@ export function setFakeConsumerAssignment(
 	nextAssignment = assignment;
 }
 
+/**
+ * How the next call(s) to `fetchTopicMetadata` answer, one entry consumed per
+ * call across every fake admin. A test that needs a missing topic (or an
+ * inconclusive check) queues outcomes before acting; once the queue is empty,
+ * calls resolve with a healthy topic.
+ */
+const metadataOutcomeQueue: Array<() => Promise<unknown>> = [];
+
+/** How the next fake admin answers `disconnect()`. */
+let nextDisconnectError: Error | undefined;
+
+/**
+ * Makes the next `fetchTopicMetadata` call reject with `error`. Queue it more
+ * than once to fail several calls in a row, e.g. across a retry.
+ */
+export function failNextTopicMetadata(error: Error): void {
+	metadataOutcomeQueue.push(async () => {
+		throw error;
+	});
+}
+
+/** Makes the next admin's `disconnect()` reject with `error`. */
+export function failNextAdminDisconnect(error: Error): void {
+	nextDisconnectError = error;
+}
+
+/** An error shaped like the library's rejection for a topic the broker does not know. */
+export function unknownTopicError(): Error & { code: number } {
+	return Object.assign(new Error('Broker: Unknown topic or partition'), {
+		name: 'KafkaJSProtocolError',
+		code: UNKNOWN_TOPIC_OR_PART,
+	});
+}
+
+/** An error shaped like the library's rejection for a topic name the broker rejects outright. */
+export function invalidTopicNameError(): Error & { code: number } {
+	return Object.assign(new Error('Broker: Invalid topic'), {
+		name: 'KafkaJSProtocolError',
+		code: TOPIC_EXCEPTION,
+	});
+}
+
+function createFakeAdmin(): FakeAdmin {
+	const disconnectError = nextDisconnectError;
+	nextDisconnectError = undefined;
+
+	const admin: FakeAdmin = {
+		connect: vi.fn(async () => {}),
+		fetchTopicMetadata: vi.fn(async () => {
+			const outcome = metadataOutcomeQueue.shift();
+			return outcome ? await outcome() : [{ name: 'test-topic', partitions: [{ partitionId: 0 }] }];
+		}),
+		disconnect: vi.fn(async () => {
+			if (disconnectError) throw disconnectError;
+		}),
+	};
+
+	admins.push(admin);
+	return admin;
+}
+
 function createFakeConsumer(config: KafkaJS.ConsumerConstructorConfig): FakeConsumer {
 	let eachBatch: EachBatchHandler | undefined;
 
@@ -135,7 +210,7 @@ function fakeKafkaClient(config?: KafkaJS.CommonConstructorConfig) {
 			disconnect: vi.fn(),
 		})),
 		consumer: vi.fn(createFakeConsumer),
-		admin: vi.fn(),
+		admin: vi.fn(createFakeAdmin),
 	};
 }
 
@@ -147,6 +222,10 @@ export function confluentKafkaModuleMock(): { readonly KafkaJS: unknown } {
 			return {
 				Kafka: vi.fn(fakeKafkaClient),
 				logLevel: { NOTHING: 0, ERROR: 1, WARN: 2, INFO: 3, DEBUG: 4 },
+				ErrorCodes: {
+					ERR_UNKNOWN_TOPIC_OR_PART: UNKNOWN_TOPIC_OR_PART,
+					ERR_TOPIC_EXCEPTION: TOPIC_EXCEPTION,
+				},
 			};
 		},
 	};
@@ -165,6 +244,11 @@ export function getFakeConsumers(): FakeConsumer[] {
 	return consumers;
 }
 
+/** Admin clients created through the fake, in creation order. */
+export function getFakeAdmins(): FakeAdmin[] {
+	return admins;
+}
+
 /** Configs passed to the fake `Kafka` constructor, in creation order. */
 export function getFakeClientConfigs(): KafkaJS.CommonConstructorConfig[] {
 	return clientConfigs;
@@ -173,6 +257,9 @@ export function getFakeClientConfigs(): KafkaJS.CommonConstructorConfig[] {
 /** Clears the recorded consumers and client configs (not the access count). */
 export function resetConfluentKafkaRecordings(): void {
 	consumers.length = 0;
+	admins.length = 0;
 	clientConfigs.length = 0;
 	nextAssignment = joinedAssignment;
+	metadataOutcomeQueue.length = 0;
+	nextDisconnectError = undefined;
 }

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

@@ -16,9 +16,12 @@ import {
 } from '../../v2/TriggerSettings';
 import {
 	confluentKafkaModuleMock,
+	failNextTopicMetadata,
+	getFakeAdmins,
 	getFakeConsumers,
 	resetConfluentKafkaRecordings,
 	setFakeConsumerAssignment,
+	unknownTopicError,
 	type FakeConsumer,
 } from '../mocks/confluent-kafka';
 
@@ -508,6 +511,66 @@ describe('KafkaTriggerV2 Node', () => {
 		expect(consumer.disconnect).toHaveBeenCalled();
 	});
 
+	describe('a topic that does not exist', () => {
+		it('fails activation instead of publishing a workflow that consumes nothing', async () => {
+			failNextTopicMetadata(unknownTopicError());
+			failNextTopicMetadata(unknownTopicError());
+
+			await expect(startTrigger('v2-missing-topic')).rejects.toThrow(
+				'Kafka topic "test-topic" does not exist',
+			);
+		});
+
+		it('never starts a consumer, so nothing is left connected', async () => {
+			failNextTopicMetadata(unknownTopicError());
+			failNextTopicMetadata(unknownTopicError());
+
+			await expect(startTrigger('v2-missing-topic-no-consumer')).rejects.toThrow();
+
+			expect(getFakeConsumers()).toHaveLength(0);
+			expect(consumeTopicSpy).not.toHaveBeenCalled();
+		});
+
+		it('fails a manual test run too, rather than listening forever', async () => {
+			// A manual run starts the consumer from `manualTriggerFunction`, so the
+			// check runs there rather than during trigger() itself.
+			const { manualTriggerFunction } = await testTriggerNode(new KafkaTriggerV2(baseDescription), {
+				mode: 'manual',
+				node: {
+					parameters: {
+						topic: 'test-topic',
+						groupId: 'v2-missing-topic-manual',
+						useSchemaRegistry: false,
+					},
+				},
+				credential,
+			});
+
+			failNextTopicMetadata(unknownTopicError());
+			failNextTopicMetadata(unknownTopicError());
+
+			await expect(manualTriggerFunction?.()).rejects.toThrow(
+				'Kafka topic "test-topic" does not exist',
+			);
+			expect(getFakeConsumers()).toHaveLength(0);
+		});
+
+		it('starts normally when the broker confirms the topic', async () => {
+			const { close } = await startTrigger('v2-topic-exists');
+
+			const admin = getFakeAdmins().at(-1);
+			expect(admin?.fetchTopicMetadata).toHaveBeenCalledWith({
+				topics: ['test-topic'],
+				timeout: 3_000,
+			});
+			// Checked with a throwaway client that does not outlive the check.
+			expect(admin?.disconnect).toHaveBeenCalledTimes(1);
+			expect(getFakeConsumers()).toHaveLength(1);
+
+			await close();
+		});
+	});
+
 	describe('Batch Size', () => {
 		it('starts one execution per message by default, as v1 does', async () => {
 			const { emit } = await startTrigger('v2-batch-default');

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

@@ -33,6 +33,7 @@ import type { KafkaCredentials } from '../../utils';
 import { KafkaTriggerV1 } from '../../v1/KafkaTriggerV1.node';
 import { consumeTopic, type KafkaConsumerHandle } from '../../v2/consumer/ConsumeTopic';
 import { createMessageParser } from '../../v2/consumer/MessageParser';
+import { assertTopicExists } from '../../v2/transport/admin';
 import { createKafkaClient } from '../../v2/transport/client';
 import { createKafkaConsumer } from '../../v2/transport/consumer';
 import { createLibraryLogger } from '../../v2/transport/LibraryLogger';
@@ -405,6 +406,41 @@ describe('library logging against a real broker', () => {
 	}, 60_000);
 });
 
+describe('a missing topic against a real broker', () => {
+	it('refuses activation, naming the topic', async () => {
+		const topic = uniqueTopic('never-created');
+
+		await expect(assertTopicExists(credentials, topic, logger)).rejects.toThrow(
+			`Kafka topic "${topic}" does not exist`,
+		);
+	});
+
+	it('lets an existing topic through', async () => {
+		const topic = uniqueTopic('exists');
+		await createTopic(topic);
+
+		await expect(assertTopicExists(credentials, topic, logger)).resolves.toBeUndefined();
+	});
+
+	it('is the only thing that catches it: subscribe and run both resolve', async () => {
+		// The behaviour the check exists for. Without it the trigger reaches "started"
+		// on a topic the broker does not have, and only a swallowed retry log says so.
+		const topic = uniqueTopic('silent-start');
+		const consumer = await createKafkaConsumer(credentials, {
+			groupId: `${topic}-group`,
+			fromBeginning: true,
+		});
+
+		try {
+			await consumer.connect();
+			await expect(consumer.subscribe({ topics: [topic] })).resolves.toBeUndefined();
+			await expect(consumer.run({ eachBatch: async () => {} })).resolves.toBeUndefined();
+		} finally {
+			await withDeadline(consumer.disconnect(), 30_000, 'disconnect').catch(() => undefined);
+		}
+	}, 60_000);
+});
+
 describe('version 1 and version 2 item parity', () => {
 	const baseDescription: INodeTypeBaseDescription = {
 		displayName: 'Kafka Trigger',

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

@@ -0,0 +1,176 @@
+import type { Logger } from 'n8n-workflow';
+import { UserError } from 'n8n-workflow';
+import { mock } from 'vitest-mock-extended';
+
+import type { KafkaCredentials } from '../../../utils';
+import { assertTopicExists } from '../../../v2/transport/admin';
+import {
+	confluentKafkaModuleMock,
+	failNextAdminDisconnect,
+	failNextTopicMetadata,
+	getFakeAdmins,
+	getFakeClientConfigs,
+	invalidTopicNameError,
+	resetConfluentKafkaRecordings,
+	unknownTopicError,
+} from '../../mocks/confluent-kafka';
+
+vi.mock('@confluentinc/kafka-javascript', () => confluentKafkaModuleMock());
+vi.mock('@n8n/utils/sleep', () => ({ sleep: vi.fn(async () => {}) }));
+
+const credentials: KafkaCredentials = {
+	clientId: 'n8n-test',
+	brokers: 'localhost:9092',
+	ssl: false,
+	authentication: false,
+};
+
+const logger = mock<Logger>();
+
+beforeEach(() => {
+	resetConfluentKafkaRecordings();
+	vi.clearAllMocks();
+});
+
+const lastAdmin = () => {
+	const admin = getFakeAdmins().at(-1);
+	if (!admin) throw new Error('the fake recorded no admin client');
+	return admin;
+};
+
+describe('assertTopicExists', () => {
+	it('asks the broker only about the topic the trigger will subscribe to', async () => {
+		await assertTopicExists(credentials, 'my-topic', logger);
+
+		expect(lastAdmin().fetchTopicMetadata).toHaveBeenCalledWith({
+			topics: ['my-topic'],
+			timeout: 3_000,
+		});
+	});
+
+	it('builds the admin client from the same converted credential as the consumer', async () => {
+		await assertTopicExists(credentials, 'my-topic', logger);
+
+		expect(getFakeClientConfigs()).toStrictEqual([
+			{
+				kafkaJS: {
+					brokers: ['localhost:9092'],
+					clientId: 'n8n-test',
+					ssl: false,
+					logLevel: 1,
+				},
+			},
+		]);
+	});
+
+	it('resolves quietly when the topic exists', async () => {
+		await expect(assertTopicExists(credentials, 'my-topic', logger)).resolves.toBeUndefined();
+
+		expect(logger.warn).not.toHaveBeenCalled();
+	});
+
+	it('fails when the broker still does not know the topic after a retry, naming it and how to recover', async () => {
+		failNextTopicMetadata(unknownTopicError());
+		failNextTopicMetadata(unknownTopicError());
+
+		const assertion = assertTopicExists(credentials, 'missing-topic', logger);
+
+		await expect(assertion).rejects.toThrow(UserError);
+		await expect(assertion).rejects.toThrow('Kafka topic "missing-topic" does not exist');
+		expect(lastAdmin().fetchTopicMetadata).toHaveBeenCalledTimes(2);
+	});
+
+	it('recovers when the second answer finds the topic, since one unknown-topic answer can be stale', async () => {
+		// The broker can call a topic unknown while metadata is still propagating
+		// (e.g. just created); the admin path retries once before it verdicts.
+		failNextTopicMetadata(unknownTopicError());
+
+		await expect(assertTopicExists(credentials, 'my-topic', logger)).resolves.toBeUndefined();
+
+		expect(lastAdmin().fetchTopicMetadata).toHaveBeenCalledTimes(2);
+		expect(logger.warn).not.toHaveBeenCalled();
+	});
+
+	it('puts the fix in the message, which is the only part a failed publish shows', async () => {
+		// n8n's activation path drops the description of a NodeOperationError, so
+		// guidance that lives only there never reaches the user.
+		failNextTopicMetadata(unknownTopicError());
+		failNextTopicMetadata(unknownTopicError());
+
+		await expect(assertTopicExists(credentials, 'missing-topic', logger)).rejects.toThrow(
+			'Create the topic on the broker, or correct the Topic field, then publish the workflow again',
+		);
+	});
+
+	it('proceeds on an inconclusive check rather than blocking activation', async () => {
+		// A broker that cannot be reached is not proof of a missing topic, and the
+		// consumer's own connect reports it with a better message a moment later.
+		failNextTopicMetadata(new Error('Local: Broker transport failure'));
+
+		await expect(assertTopicExists(credentials, 'my-topic', logger)).resolves.toBeUndefined();
+
+		expect(logger.warn).toHaveBeenCalledWith(
+			'Kafka topic could not be verified before starting the consumer',
+			expect.objectContaining({ topic: 'my-topic' }),
+		);
+	});
+
+	it('skips the check for a pattern topic, without asking the broker at all', async () => {
+		// A Topic of `^orders-.*` is a working pattern subscription on v2. Asking
+		// the broker would only get "invalid topic" (17), always inconclusive, so
+		// it never changes the outcome — it would just log a warning on every
+		// activation of an otherwise healthy pattern.
+		await expect(assertTopicExists(credentials, '^orders-.*', logger)).resolves.toBeUndefined();
+
+		expect(getFakeAdmins()).toHaveLength(0);
+		expect(logger.warn).not.toHaveBeenCalled();
+	});
+
+	it('fails activation when the broker rejects the name outright, since no wait fixes that', async () => {
+		failNextTopicMetadata(invalidTopicNameError());
+
+		const assertion = assertTopicExists(credentials, 'bad topic ', logger);
+
+		await expect(assertion).rejects.toThrow(UserError);
+		await expect(assertion).rejects.toThrow('is not a valid Kafka topic name');
+		// Deterministic, not propagation-dependent: unlike "unknown topic", one
+		// answer is the verdict, so there is no retry to ask twice for.
+		expect(lastAdmin().fetchTopicMetadata).toHaveBeenCalledTimes(1);
+	});
+
+	it('does not treat some other error code as a missing topic', async () => {
+		failNextTopicMetadata(
+			Object.assign(new Error('Broker: Group authorization failed'), { code: 30 }),
+		);
+
+		await expect(assertTopicExists(credentials, 'my-topic', logger)).resolves.toBeUndefined();
+	});
+
+	it('disconnects the admin client on every path', async () => {
+		await assertTopicExists(credentials, 'my-topic', logger);
+		expect(lastAdmin().disconnect).toHaveBeenCalledTimes(1);
+
+		failNextTopicMetadata(unknownTopicError());
+		failNextTopicMetadata(unknownTopicError());
+		await expect(assertTopicExists(credentials, 'missing-topic', logger)).rejects.toThrow(
+			UserError,
+		);
+		expect(lastAdmin().disconnect).toHaveBeenCalledTimes(1);
+	});
+
+	it('reports the missing topic even if the admin disconnect then fails', async () => {
+		failNextTopicMetadata(unknownTopicError());
+		failNextTopicMetadata(unknownTopicError());
+		failNextAdminDisconnect(new Error('disconnect timed out'));
+
+		await expect(assertTopicExists(credentials, 'missing-topic', logger)).rejects.toThrow(
+			'Kafka topic "missing-topic" does not exist',
+		);
+	});
+
+	it('works without a logger', async () => {
+		failNextTopicMetadata(new Error('Local: Broker transport failure'));
+
+		await expect(assertTopicExists(credentials, 'my-topic')).resolves.toBeUndefined();
+	});
+});

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

@@ -12,7 +12,7 @@ import { setSchemaRegistry, type KafkaCredentials } from '../utils';
 import { consumeTopic, createDataEmitter, createMessageParser } from './consumer';
 import type { KafkaConsumerHandle } from './consumer';
 import { versionDescription } from './KafkaTriggerV2Description';
-import { createKafkaConsumer } from './transport';
+import { assertTopicExists, createKafkaConsumer } from './transport';
 import { explainManualRunGroupDenial, getSettings } from './TriggerSettings';
 
 export class KafkaTriggerV2 implements INodeType {
@@ -72,6 +72,10 @@ export class KafkaTriggerV2 implements INodeType {
 			void startupFailure.catch(() => {});
 
 			try {
+				// Before the consumer, so a missing topic fails activation instead of
+				// leaving a Published workflow that silently consumes nothing.
+				await assertTopicExists(credentials, settings.topic, this.logger);
+
 				const consumer = await createKafkaConsumer(credentials, settings.consumer, {
 					logger: this.logger,
 					// v1 routes non-restartable consumer crashes to emitError so n8n

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

@@ -0,0 +1,117 @@
+import { sleep } from '@n8n/utils/sleep';
+import type { Logger } from 'n8n-workflow';
+import { UserError } from 'n8n-workflow';
+
+import { createKafkaClient, getKafkaLibrary } from './client';
+import { createLibraryLogger } from './LibraryLogger';
+import type { KafkaCredentials } from '../../utils';
+
+/**
+ * Bounds the metadata request so an unreachable broker can't stall
+ * activation: `admin.connect()` resolves without reaching the broker, so the
+ * whole wait for a dead broker lands here.
+ */
+const METADATA_TIMEOUT_MS = 3_000;
+
+/**
+ * "Unknown topic" is retriable, not final: a topic just created can still be
+ * reported unknown while metadata propagates, and the admin path (unlike v1's
+ * kafkajs consumer) does not retry that on its own. One retry, once, is
+ * enough to tell a stale answer from a real one without adding much delay.
+ */
+const UNKNOWN_TOPIC_RETRY_DELAY_MS = 500;
+
+/**
+ * Fails activation when the topic does not exist, since neither `subscribe()`
+ * nor `run()` reject for a missing topic: the workflow would otherwise show
+ * as Published while silently consuming nothing until the next broker
+ * metadata refresh, 5 minutes out by default.
+ *
+ * Two verdicts block activation: "unknown topic", and "invalid topic" for a
+ * non-pattern name. Both are as final as each other — a name Kafka rejects
+ * (trailing space, a comma-joined list pasted as one topic, over 249 chars)
+ * never becomes valid by waiting, so letting it through would activate into
+ * the same silent, nothing-consumed state this check exists to prevent.
+ * Anything else is inconclusive and left for the consumer's own connect to
+ * report.
+ *
+ * Skips the check entirely for a pattern topic (leading `^`): the broker
+ * answers those with "invalid topic" too, but that verdict only means
+ * "not a valid literal name," which a pattern was never meant to be — asking
+ * would only log a misleading warning on every activation of a healthy
+ * pattern.
+ * @param credentials - The decrypted Kafka credential
+ * @param topic - The topic the trigger is about to subscribe to
+ * @param logger - Records an inconclusive check, which is not an error
+ */
+export async function assertTopicExists(
+	credentials: KafkaCredentials,
+	topic: string,
+	logger?: Logger,
+): Promise<void> {
+	if (topic.startsWith('^')) return;
+
+	const { ErrorCodes } = await getKafkaLibrary();
+	const kafka = await createKafkaClient(credentials);
+	const admin = kafka.admin({
+		// Without this the library's own logger writes ERROR-and-above straight
+		// to stdout, bypassing n8n's logger; there is no fatal-error handler here
+		// since this admin client is short-lived and has no run loop to abort.
+		...(logger ? { kafkaJS: { logger: createLibraryLogger(logger) } } : {}),
+	});
+
+	try {
+		await admin.connect();
+
+		try {
+			await admin.fetchTopicMetadata({ topics: [topic], timeout: METADATA_TIMEOUT_MS });
+			return;
+		} catch (error) {
+			if (!hasErrorCode(error, ErrorCodes.ERR_UNKNOWN_TOPIC_OR_PART)) throw error;
+			await sleep(UNKNOWN_TOPIC_RETRY_DELAY_MS);
+			await admin.fetchTopicMetadata({ topics: [topic], timeout: METADATA_TIMEOUT_MS });
+		}
+	} catch (error) {
+		if (hasErrorCode(error, ErrorCodes.ERR_UNKNOWN_TOPIC_OR_PART)) {
+			// The description is dropped on a failed publish, so the fix instruction
+			// must be in the message; the description still renders elsewhere.
+			throw new UserError(
+				`Kafka topic "${topic}" does not exist. Create the topic on the broker, or correct the Topic field, then publish the workflow again.`,
+				{
+					level: 'warning',
+					description:
+						'Publishing anyway would leave the workflow showing as published while consuming nothing, because a topic created later is only picked up at the next broker metadata refresh, minutes away.',
+					cause: error instanceof Error ? error : undefined,
+				},
+			);
+		}
+
+		// A pattern topic never reaches here (it returns above), so this is always
+		// a literal name Kafka's own naming rules reject outright, not a
+		// propagation delay — no amount of waiting fixes it.
+		if (hasErrorCode(error, ErrorCodes.ERR_TOPIC_EXCEPTION)) {
+			throw new UserError(
+				`Kafka topic "${topic}" is not a valid Kafka topic name. Correct the Topic field, then publish the workflow again.`,
+				{ level: 'warning', cause: error instanceof Error ? error : undefined },
+			);
+		}
+
+		logger?.warn('Kafka topic could not be verified before starting the consumer', {
+			topic,
+			error,
+		});
+	} finally {
+		// This call is synchronous.
+		await admin.disconnect().catch(() => {});
+	}
+}
+
+/**
+ * Whether the broker's error matches the given code. Checked by code, not
+ * message: the admin path preserves the broker's error code, unlike the
+ * consumer's log stream.
+ */
+function hasErrorCode(error: unknown, code: number): boolean {
+	if (typeof error !== 'object' || error === null || !('code' in error)) return false;
+	return error.code === code;
+}

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

@@ -1,5 +1,6 @@
 export { toKafkaJSConfig } from './config';
 export { getKafkaLibrary, createKafkaClient } from './client';
+export { assertTopicExists } from './admin';
 export { createKafkaProducer, type KafkaProducerOptions } from './producer';
 export {
 	createKafkaConsumer,