diff --git a/packages/nodes-base/nodes/Kafka/test/mocks/confluent-kafka.ts b/packages/nodes-base/nodes/Kafka/test/mocks/confluent-kafka.ts index c8cae062d78..b264b23bc76 100644 --- a/packages/nodes-base/nodes/Kafka/test/mocks/confluent-kafka.ts +++ b/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> = []; + +/** 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; } diff --git a/packages/nodes-base/nodes/Kafka/test/v2/KafkaTriggerV2.node.test.ts b/packages/nodes-base/nodes/Kafka/test/v2/KafkaTriggerV2.node.test.ts index c4970992845..0ba0e72d465 100644 --- a/packages/nodes-base/nodes/Kafka/test/v2/KafkaTriggerV2.node.test.ts +++ b/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'); diff --git a/packages/nodes-base/nodes/Kafka/test/v2/kafka.integration.test.ts b/packages/nodes-base/nodes/Kafka/test/v2/kafka.integration.test.ts index d33cf631b2b..d7acfbd67a7 100644 --- a/packages/nodes-base/nodes/Kafka/test/v2/kafka.integration.test.ts +++ b/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', diff --git a/packages/nodes-base/nodes/Kafka/test/v2/transport/admin.test.ts b/packages/nodes-base/nodes/Kafka/test/v2/transport/admin.test.ts new file mode 100644 index 00000000000..78da5a7d3ec --- /dev/null +++ b/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(); + +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(); + }); +}); diff --git a/packages/nodes-base/nodes/Kafka/v2/KafkaTriggerV2.node.ts b/packages/nodes-base/nodes/Kafka/v2/KafkaTriggerV2.node.ts index 3bb89871160..af446114235 100644 --- a/packages/nodes-base/nodes/Kafka/v2/KafkaTriggerV2.node.ts +++ b/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 diff --git a/packages/nodes-base/nodes/Kafka/v2/transport/admin.ts b/packages/nodes-base/nodes/Kafka/v2/transport/admin.ts new file mode 100644 index 00000000000..31f32182e9f --- /dev/null +++ b/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 { + 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; +} diff --git a/packages/nodes-base/nodes/Kafka/v2/transport/index.ts b/packages/nodes-base/nodes/Kafka/v2/transport/index.ts index eb4449505a5..51c427824a1 100644 --- a/packages/nodes-base/nodes/Kafka/v2/transport/index.ts +++ b/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,