feat(Kafka Node): Add version 2 to send messages via the new Kafka library (#35450)

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

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

@@ -2,6 +2,7 @@ import type { INodeTypeBaseDescription, IVersionedNodeType } from 'n8n-workflow'
 import { VersionedNodeType } from 'n8n-workflow';
 
 import { KafkaV1 } from './v1/KafkaV1.node';
+import { KafkaV2 } from './v2/KafkaV2.node';
 
 export class Kafka extends VersionedNodeType {
 	constructor() {
@@ -16,6 +17,7 @@ export class Kafka extends VersionedNodeType {
 
 		const nodeVersions: IVersionedNodeType['nodeVersions'] = {
 			1: new KafkaV1(baseDescription),
+			2: new KafkaV2(baseDescription),
 		};
 
 		super(nodeVersions, baseDescription);

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

@@ -11,6 +11,12 @@ import { mock } from 'vitest-mock-extended';
 
 import { Kafka } from '../Kafka.node';
 import { KafkaV1 } from '../v1/KafkaV1.node';
+import { KafkaV2 } from '../v2/KafkaV2.node';
+import {
+	confluentKafkaModuleMock,
+	getConfluentKafkaAccessCount,
+	resetConfluentKafkaAccessCount,
+} from './mocks/confluent-kafka';
 
 // The node is imported directly (through vite) so vi.mock can intercept its
 // `kafkajs` / `@kafkajs/confluent-schema-registry` imports. NodeTestHarness can't
@@ -91,6 +97,11 @@ vi.mock('@kafkajs/confluent-schema-registry', () => ({
 	}),
 }));
 
+// v1 must never load the new library — the ESLint import restrictions guard the
+// static-import side; this covers the runtime side (e.g. a dynamic import added
+// by mistake down the line).
+vi.mock('@confluentinc/kafka-javascript', () => confluentKafkaModuleMock());
+
 const defaultKafkaCredentials: IDataObject = {
 	brokers: 'localhost:9092',
 	clientId: 'test-client',
@@ -145,6 +156,24 @@ const schemaRegistryCredential = {
 describe('Kafka Node', () => {
 	beforeEach(() => {
 		vi.clearAllMocks();
+		resetConfluentKafkaAccessCount();
+	});
+
+	test('never loads the new confluent-kafka-javascript library', async () => {
+		const params: IDataObject = {
+			options: { acks: true, compression: true, timeout: 1000 },
+			sendInputData: true,
+			useSchemaRegistry: false,
+			topic: 'test-topic',
+			jsonParameters: false,
+			useKey: false,
+			headersUi: {},
+		};
+		const items: INodeExecutionData[] = [{ json: { name: 'item' } }];
+
+		await new KafkaV1(baseDescription).execute.call(createExecuteFunctions(params, items));
+
+		expect(getConfluentKafkaAccessCount()).toBe(0);
 	});
 
 	test('publishes input data as messages with key, headers and options', async () => {
@@ -402,6 +431,30 @@ describe('Kafka (versioned entry point)', () => {
 		expect(kafka.nodeVersions[1]).toBeInstanceOf(KafkaV1);
 	});
 
+	it('should expose version 2 as KafkaV2', () => {
+		expect(kafka.nodeVersions[2]).toBeInstanceOf(KafkaV2);
+	});
+
+	// One credential test per credential type, not per node version. Adding
+	// `methods.credentialTest.kafkaConnectionTest` to v2 wouldn't add a second test — it is
+	// resolved newest-version-first, so it would take over v1's test for everyone. Until it
+	// moves, the test exercises v1's kafkajs path while v2 connects through librdkafka.
+	it('should leave the kafka credential test to v1', () => {
+		const v2 = kafka.nodeVersions[2];
+
+		expect(v2.methods?.credentialTest).toBeUndefined();
+		expect(v2.description.credentials?.find((c) => c.name === 'kafka')?.testedBy).toBeUndefined();
+		expect(kafka.nodeVersions[1].methods?.credentialTest).toHaveProperty('kafkaConnectionTest');
+	});
+
+	it('should resolve v1 by default', () => {
+		expect(kafka.getNodeType()).toBeInstanceOf(KafkaV1);
+	});
+
+	it('should resolve v2 when requested', () => {
+		expect(kafka.getNodeType(2)).toBeInstanceOf(KafkaV2);
+	});
+
 	it('should have defaultVersion set to 1', () => {
 		expect(kafka.description.defaultVersion).toBe(1);
 	});

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

@@ -8,14 +8,20 @@ export function confluentKafkaModuleMock(): { readonly KafkaJS: unknown } {
 		get KafkaJS() {
 			accessCount += 1;
 			return {
-				Kafka: vi.fn().mockImplementation((config?: unknown) => ({
-					config,
-					connect: vi.fn(),
-					disconnect: vi.fn(),
-					producer: vi.fn(),
-					consumer: vi.fn(),
-					admin: vi.fn(),
-				})),
+				Kafka: vi.fn().mockImplementation(function (config?: unknown) {
+					return {
+						config,
+						connect: vi.fn(),
+						disconnect: vi.fn(),
+						producer: vi.fn(() => ({
+							connect: vi.fn(),
+							sendBatch: vi.fn(),
+							disconnect: vi.fn(),
+						})),
+						consumer: vi.fn(),
+						admin: vi.fn(),
+					};
+				}),
 				logLevel: { NOTHING: 0, ERROR: 1, WARN: 2, INFO: 3, DEBUG: 4 },
 			};
 		},

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

@@ -0,0 +1,595 @@
+import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
+import type {
+	IDataObject,
+	IExecuteFunctions,
+	INode,
+	INodeExecutionData,
+	INodeTypeBaseDescription,
+} from 'n8n-workflow';
+import { NodeOperationError } from 'n8n-workflow';
+import { mock } from 'vitest-mock-extended';
+
+import { KafkaV2 } from '../../v2/KafkaV2.node';
+import { getKafkaLibrary } from '../../v2/transport/client';
+import { confluentKafkaModuleMock } from '../mocks/confluent-kafka';
+
+// Same reasoning as the v1 test file: the node is imported directly (through vite)
+// so vi.mock can intercept its library imports; NodeTestHarness loads from dist via
+// require(), where vi.mock can't reach it.
+const baseDescription: INodeTypeBaseDescription = {
+	displayName: 'Kafka',
+	name: 'kafka',
+	group: ['transform'],
+	description: 'Sends messages to a Kafka topic',
+};
+
+const {
+	kafkajsLoadCount,
+	mockProducerConnect,
+	mockProducerSendBatch,
+	mockProducerDisconnect,
+	mockProducerFactory,
+	mockRegistryEncode,
+	mockRegistryGetLatestSchemaId,
+} = vi.hoisted(() => {
+	const kafkajsLoadCount = { value: 0 };
+
+	const mockProducerConnect = vi.fn(async () => {});
+	const mockProducerSendBatch = vi.fn(async () => [] as unknown[]);
+	const mockProducerDisconnect = vi.fn(async () => {});
+	const mockProducerFactory = vi.fn(() => ({
+		connect: mockProducerConnect,
+		sendBatch: mockProducerSendBatch,
+		disconnect: mockProducerDisconnect,
+	}));
+
+	const mockRegistryEncode = vi.fn(async (_id: number, input: unknown) =>
+		Buffer.from(JSON.stringify(input)),
+	);
+	const mockRegistryGetLatestSchemaId = vi.fn(async (eventName: string) => {
+		if (eventName === 'failing-event-name') {
+			throw new Error('Subject not found');
+		}
+		return 1;
+	});
+
+	return {
+		kafkajsLoadCount,
+		mockProducerConnect,
+		mockProducerSendBatch,
+		mockProducerDisconnect,
+		mockProducerFactory,
+		mockRegistryEncode,
+		mockRegistryGetLatestSchemaId,
+	};
+});
+
+vi.mock('@confluentinc/kafka-javascript', () => confluentKafkaModuleMock());
+
+// Counts module loads, not property reads: `kafkajs` reaching v2 would do so through
+// a static value import somewhere in its graph (e.g. the shared `utils.ts`), which
+// resolves once at import time — so the counter is never reset between tests.
+vi.mock('kafkajs', () => {
+	kafkajsLoadCount.value += 1;
+	return { logLevel: { NOTHING: 0, ERROR: 1, WARN: 2, INFO: 3, DEBUG: 4 } };
+});
+
+vi.mock('@kafkajs/confluent-schema-registry', () => ({
+	SchemaRegistry: vi.fn(function () {
+		return {
+			getLatestSchemaId: mockRegistryGetLatestSchemaId,
+			encode: mockRegistryEncode,
+		};
+	}),
+}));
+
+const defaultKafkaCredentials: IDataObject = {
+	brokers: 'localhost:9092',
+	clientId: 'test-client',
+	ssl: false,
+	authentication: false,
+};
+
+// A param value may be a `(index) => value` function, so the item index the node
+// reads a per-item parameter at is actually observable.
+type NodeParams = Record<string, unknown>;
+
+const mockLoggerWarn = vi.fn();
+
+function createExecuteFunctions(
+	params: NodeParams,
+	items: INodeExecutionData[],
+	options: {
+		schemaRegistryCredential?: IDataObject;
+		continueOnFail?: boolean;
+	} = {},
+) {
+	const { schemaRegistryCredential, continueOnFail = false } = options;
+
+	const node = mock<INode>({
+		name: 'Kafka',
+		credentials: schemaRegistryCredential
+			? { schemaRegistryApi: { id: 'wW0eW1iZK9d3Yz2g', name: 'Schema Registry account' } }
+			: undefined,
+	});
+
+	return mock<IExecuteFunctions>({
+		getInputData: () => items,
+		getNode: () => node,
+		logger: mock<IExecuteFunctions['logger']>({ warn: mockLoggerWarn }),
+		getNodeParameter: ((name: string, index: number, fallback?: unknown) => {
+			if (!(name in params)) return fallback;
+			const value = params[name];
+			return typeof value === 'function' ? (value as (i: number) => unknown)(index) : value;
+		}) as IExecuteFunctions['getNodeParameter'],
+		getCredentials: (async (type: string) =>
+			type === 'schemaRegistryApi'
+				? schemaRegistryCredential
+				: defaultKafkaCredentials) as IExecuteFunctions['getCredentials'],
+		continueOnFail: () => continueOnFail,
+		helpers: {
+			returnJsonArray: (data: IDataObject | IDataObject[]) =>
+				(Array.isArray(data) ? data : [data]).map((json) => ({ json })),
+			constructExecutionMetaData: (data: INodeExecutionData[]) => data,
+		} as unknown as IExecuteFunctions['helpers'],
+	});
+}
+
+const schemaRegistryCredential = {
+	url: 'https://cred-kafka-registry.local',
+	authentication: 'basicAuth',
+	username: 'registry-user',
+	password: 'registry-password',
+};
+
+describe('KafkaV2 Node', () => {
+	beforeEach(async () => {
+		vi.clearAllMocks();
+		// The shared fake's own `Kafka` mock returns a fresh, uninstrumented stub producer
+		// per instance — fine for the transport-level tests, but this file needs the same
+		// controllable producer across calls, so the constructor implementation is
+		// overridden (still through the shared fake's `vi.mock`, so lazy-loading semantics
+		// stay real) to return `mockProducerFactory`'s producer instead.
+		const { Kafka } = await getKafkaLibrary();
+		vi.mocked(Kafka).mockImplementation(function (config?: unknown) {
+			return {
+				config,
+				connect: vi.fn(),
+				disconnect: vi.fn(),
+				producer: mockProducerFactory,
+				consumer: vi.fn(),
+				admin: vi.fn(),
+			};
+		});
+	});
+
+	test('never loads the v1 kafkajs library', async () => {
+		const params: IDataObject = {
+			options: {},
+			sendInputData: true,
+			useSchemaRegistry: false,
+			topic: 'test-topic',
+			jsonParameters: false,
+			useKey: false,
+			headersUi: {},
+		};
+		const items: INodeExecutionData[] = [{ json: { name: 'item' } }];
+
+		await new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, items));
+
+		expect(kafkajsLoadCount.value).toBe(0);
+	});
+
+	test('publishes input data as messages with key and headers unchanged, acks/timeout on the producer config', async () => {
+		const params: IDataObject = {
+			options: { acks: true, timeout: 1000 },
+			sendInputData: true,
+			useSchemaRegistry: false,
+			topic: 'test-topic',
+			jsonParameters: false,
+			useKey: true,
+			key: 'messageKey',
+			headersUi: { headerValues: [{ key: 'header', value: 'value' }] },
+		};
+		const items: INodeExecutionData[] = [
+			{ json: { name: 'First item', code: 1 } },
+			{ json: { name: 'Second item', code: 2 } },
+		];
+
+		await new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, items));
+
+		expect(mockProducerFactory).toHaveBeenCalledWith({
+			kafkaJS: { acks: -1, timeout: 1000, allowAutoTopicCreation: true },
+		});
+		expect(mockProducerConnect).toHaveBeenCalledTimes(1);
+		expect(mockProducerSendBatch).toHaveBeenCalledTimes(1);
+		expect(mockProducerSendBatch).toHaveBeenCalledWith({
+			topicMessages: [
+				{
+					messages: [
+						{
+							headers: { header: 'value' },
+							key: 'messageKey',
+							value: '{"name":"First item","code":1}',
+						},
+					],
+					topic: 'test-topic',
+				},
+				{
+					messages: [
+						{
+							headers: { header: 'value' },
+							key: 'messageKey',
+							value: '{"name":"Second item","code":2}',
+						},
+					],
+					topic: 'test-topic',
+				},
+			],
+		});
+		expect(mockProducerDisconnect).toHaveBeenCalledTimes(1);
+	});
+
+	test('maps acks off to 0 and falls back to the default timeout when the option is unset', async () => {
+		const params: IDataObject = {
+			options: {},
+			sendInputData: false,
+			useSchemaRegistry: false,
+			topic: 'test-topic',
+			jsonParameters: false,
+			useKey: false,
+			message: 'plain message',
+			headersUi: {},
+		};
+		const items: INodeExecutionData[] = [{ json: {} }];
+
+		await new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, items));
+
+		expect(mockProducerFactory).toHaveBeenCalledWith({
+			kafkaJS: { acks: 0, timeout: 30000, allowAutoTopicCreation: true },
+		});
+		expect(mockProducerSendBatch).toHaveBeenCalledWith({
+			topicMessages: [
+				{
+					messages: [{ headers: {}, key: null, value: 'plain message' }],
+					topic: 'test-topic',
+				},
+			],
+		});
+	});
+
+	test('reads the topic and key per item', async () => {
+		const params: NodeParams = {
+			options: {},
+			sendInputData: true,
+			useSchemaRegistry: false,
+			topic: (i: number) => `topic-${i}`,
+			jsonParameters: false,
+			useKey: true,
+			key: (i: number) => `key-${i}`,
+			headersUi: {},
+		};
+		const items: INodeExecutionData[] = [{ json: { a: 1 } }, { json: { a: 2 } }];
+
+		await new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, items));
+
+		expect(mockProducerSendBatch).toHaveBeenCalledWith({
+			topicMessages: [
+				{ messages: [{ headers: {}, key: 'key-0', value: '{"a":1}' }], topic: 'topic-0' },
+				{ messages: [{ headers: {}, key: 'key-1', value: '{"a":2}' }], topic: 'topic-1' },
+			],
+		});
+	});
+
+	test('reports success when the broker returns no record metadata', async () => {
+		const params: NodeParams = {
+			options: {},
+			sendInputData: true,
+			useSchemaRegistry: false,
+			topic: 'test-topic',
+			jsonParameters: false,
+			useKey: false,
+			headersUi: {},
+		};
+		const items: INodeExecutionData[] = [{ json: { name: 'item' } }];
+
+		const result = await new KafkaV2(baseDescription).execute.call(
+			createExecuteFunctions(params, items),
+		);
+
+		expect(result).toEqual([[{ json: { success: true } }]]);
+	});
+
+	test('returns the broker record metadata as item data', async () => {
+		mockProducerSendBatch.mockResolvedValueOnce([{ topicName: 't', partition: 0, offset: '1' }]);
+
+		const params: NodeParams = {
+			options: {},
+			sendInputData: true,
+			useSchemaRegistry: false,
+			topic: 't',
+			jsonParameters: false,
+			useKey: false,
+			headersUi: {},
+		};
+		const items: INodeExecutionData[] = [{ json: { name: 'item' } }];
+
+		const result = await new KafkaV2(baseDescription).execute.call(
+			createExecuteFunctions(params, items),
+		);
+
+		expect(result).toEqual([[{ json: { topicName: 't', partition: 0, offset: '1' } }]]);
+	});
+
+	test('publishes a schema-registry-encoded message as the encoded bytes', async () => {
+		const params: IDataObject = {
+			options: {},
+			sendInputData: false,
+			useSchemaRegistry: true,
+			message: JSON.stringify({ foo: 'bar' }),
+			schemaRegistryUrl: 'https://test-kafka-registry.local',
+			eventName: 'test-event-name',
+			topic: 'test-topic',
+			jsonParameters: true,
+			useKey: false,
+			headerParametersJson: '{\n  "headerKey": "headerValue"\n}',
+		};
+		const items: INodeExecutionData[] = [{ json: { success: true } }];
+
+		await new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, items));
+
+		expect(SchemaRegistry).toHaveBeenCalledWith({ host: 'https://test-kafka-registry.local' });
+		expect(mockRegistryGetLatestSchemaId).toHaveBeenCalledWith('test-event-name');
+		expect(mockRegistryEncode).toHaveBeenCalledWith(1, { foo: 'bar' });
+
+		expect(mockProducerSendBatch).toHaveBeenCalledWith({
+			topicMessages: [
+				{
+					messages: [
+						{
+							headers: { headerKey: 'headerValue' },
+							key: null,
+							value: Buffer.from(JSON.stringify({ foo: 'bar' })),
+						},
+					],
+					topic: 'test-topic',
+				},
+			],
+		});
+	});
+
+	const sendParams: NodeParams = {
+		options: {},
+		sendInputData: false,
+		useSchemaRegistry: false,
+		message: 'plain message',
+		topic: 'test-topic',
+		jsonParameters: false,
+		useKey: false,
+		headersUi: {},
+	};
+
+	test('disconnects the producer even when sendBatch rejects', async () => {
+		mockProducerSendBatch.mockRejectedValueOnce(new Error('broker unreachable'));
+
+		await expect(
+			new KafkaV2(baseDescription).execute.call(createExecuteFunctions(sendParams, [{ json: {} }])),
+		).rejects.toThrow('broker unreachable');
+
+		expect(mockProducerDisconnect).toHaveBeenCalledTimes(1);
+	});
+
+	test('surfaces the send error, not the disconnect error, when both reject', async () => {
+		mockProducerSendBatch.mockRejectedValueOnce(new Error('broker unreachable'));
+		mockProducerDisconnect.mockRejectedValueOnce(new Error('disconnect failed'));
+
+		await expect(
+			new KafkaV2(baseDescription).execute.call(createExecuteFunctions(sendParams, [{ json: {} }])),
+		).rejects.toThrow('broker unreachable');
+
+		// Swallowed, but not silently: a native client that fails to disconnect leaks threads.
+		expect(mockLoggerWarn).toHaveBeenCalledWith('Kafka producer failed to disconnect', {
+			error: 'disconnect failed',
+		});
+	});
+
+	test('still returns the send result when only the disconnect rejects', async () => {
+		mockProducerSendBatch.mockResolvedValueOnce([{ topicName: 'test-topic', partition: 0 }]);
+		mockProducerDisconnect.mockRejectedValueOnce(new Error('disconnect failed'));
+
+		const result = await new KafkaV2(baseDescription).execute.call(
+			createExecuteFunctions(sendParams, [{ json: {} }]),
+		);
+
+		// The message was accepted by the broker, so a failed cleanup must not fail the item.
+		expect(result).toEqual([[{ json: { topicName: 'test-topic', partition: 0 } }]]);
+	});
+
+	test('rejects a non-string header value before the producer is created', async () => {
+		const params: NodeParams = {
+			...sendParams,
+			jsonParameters: true,
+			headerParametersJson: '{"retries":3}',
+		};
+
+		await expect(
+			new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, [{ json: {} }])),
+		).rejects.toThrow('Header "retries" must be a string');
+
+		expect(mockProducerFactory).not.toHaveBeenCalled();
+		expect(mockProducerSendBatch).not.toHaveBeenCalled();
+	});
+
+	test('attributes malformed JSON headers to the failing item', async () => {
+		const params: NodeParams = {
+			...sendParams,
+			jsonParameters: true,
+			headerParametersJson: (i: number) => (i === 1 ? 'not json' : '{"ok":"yes"}'),
+		};
+
+		const error = await new KafkaV2(baseDescription).execute
+			.call(createExecuteFunctions(params, [{ json: {} }, { json: {} }]))
+			.catch((caught: unknown) => caught);
+
+		expect(error).toBeInstanceOf(NodeOperationError);
+		expect((error as NodeOperationError).message).toBe('Headers must be valid JSON');
+		expect((error as NodeOperationError).context.itemIndex).toBe(1);
+	});
+
+	test('disconnects and propagates the connect error when connect rejects', async () => {
+		mockProducerConnect.mockRejectedValueOnce(new Error('connection refused'));
+
+		await expect(
+			new KafkaV2(baseDescription).execute.call(createExecuteFunctions(sendParams, [{ json: {} }])),
+		).rejects.toThrow('connection refused');
+
+		expect(mockProducerSendBatch).not.toHaveBeenCalled();
+		expect(mockProducerDisconnect).toHaveBeenCalledTimes(1);
+	});
+
+	test('returns the send error as item data when the node continues on fail', async () => {
+		mockProducerSendBatch.mockRejectedValueOnce(new Error('broker unreachable'));
+
+		const result = await new KafkaV2(baseDescription).execute.call(
+			createExecuteFunctions(sendParams, [{ json: {} }], { continueOnFail: true }),
+		);
+
+		expect(result).toEqual([
+			[{ json: { error: 'broker unreachable' }, pairedItem: [{ item: 0 }] }],
+		]);
+		expect(mockProducerDisconnect).toHaveBeenCalledTimes(1);
+	});
+
+	test('resolves the schema once and encodes every item', async () => {
+		const params: NodeParams = {
+			options: {},
+			sendInputData: true,
+			useSchemaRegistry: true,
+			schemaRegistryUrl: 'https://test-kafka-registry.local',
+			eventName: 'test-event-name',
+			topic: 'test-topic',
+			jsonParameters: false,
+			useKey: false,
+			headersUi: {},
+		};
+		const items: INodeExecutionData[] = [{ json: { a: 1 } }, { json: { a: 2 } }];
+
+		await new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, items));
+
+		expect(mockRegistryGetLatestSchemaId).toHaveBeenCalledTimes(1);
+		expect(mockRegistryEncode).toHaveBeenCalledTimes(2);
+		expect(mockRegistryEncode).toHaveBeenNthCalledWith(1, 1, { a: 1 });
+		expect(mockRegistryEncode).toHaveBeenNthCalledWith(2, 1, { a: 2 });
+	});
+
+	test('fails before the producer is built when the message is not valid JSON', async () => {
+		const params: NodeParams = {
+			options: {},
+			sendInputData: false,
+			useSchemaRegistry: true,
+			message: 'not-json',
+			schemaRegistryUrl: 'https://test-kafka-registry.local',
+			eventName: 'test-event-name',
+			topic: 'test-topic',
+		};
+
+		await expect(
+			new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, [{ json: {} }])),
+		).rejects.toThrow('Message is not valid JSON');
+
+		expect(mockProducerFactory).not.toHaveBeenCalled();
+	});
+
+	test('fails before the producer is built when the JSON headers are not valid JSON', async () => {
+		const params: NodeParams = {
+			options: {},
+			sendInputData: false,
+			useSchemaRegistry: false,
+			message: 'plain message',
+			topic: 'test-topic',
+			jsonParameters: true,
+			useKey: false,
+			headerParametersJson: 'not-json',
+		};
+
+		await expect(
+			new KafkaV2(baseDescription).execute.call(createExecuteFunctions(params, [{ json: {} }])),
+		).rejects.toThrow('Headers must be valid JSON');
+
+		expect(mockProducerFactory).not.toHaveBeenCalled();
+	});
+
+	test('should configure the schema registry from the selected credential', async () => {
+		const params: IDataObject = {
+			options: {},
+			sendInputData: false,
+			useSchemaRegistry: true,
+			message: JSON.stringify({ foo: 'bar' }),
+			schemaRegistryUrl: '',
+			eventName: 'test-event-name',
+			topic: 'cred-test-topic',
+			jsonParameters: true,
+			useKey: false,
+			headerParametersJson: '{\n  "headerKey": "headerValue"\n}',
+		};
+		const items: INodeExecutionData[] = [{ json: { success: true } }];
+
+		await new KafkaV2(baseDescription).execute.call(
+			createExecuteFunctions(params, items, { schemaRegistryCredential }),
+		);
+
+		expect(SchemaRegistry).toHaveBeenCalledWith({
+			host: 'https://cred-kafka-registry.local',
+			auth: { username: 'registry-user', password: 'registry-password' },
+		});
+	});
+
+	test('should fail with the generic message when the schema lookup fails', async () => {
+		const params: IDataObject = {
+			options: {},
+			sendInputData: false,
+			useSchemaRegistry: true,
+			message: '{"foo":"bar"}',
+			schemaRegistryUrl: '',
+			eventName: 'failing-event-name',
+			topic: 'error-test-topic',
+		};
+		const items: INodeExecutionData[] = [{ json: {} }];
+
+		await expect(
+			new KafkaV2(baseDescription).execute.call(
+				createExecuteFunctions(params, items, { schemaRegistryCredential }),
+			),
+		).rejects.toThrow('Verify your Schema Registry configuration');
+
+		expect(mockProducerFactory).not.toHaveBeenCalled();
+	});
+
+	test('should return the error as item data when the node continues on fail', async () => {
+		const params: IDataObject = {
+			options: {},
+			sendInputData: false,
+			useSchemaRegistry: true,
+			message: '{"foo":"bar"}',
+			schemaRegistryUrl: '',
+			eventName: 'test-event-name',
+			topic: 'error-test-topic',
+		};
+		const items: INodeExecutionData[] = [{ json: {} }];
+
+		const result = await new KafkaV2(baseDescription).execute.call(
+			createExecuteFunctions(params, items, {
+				schemaRegistryCredential: { ...schemaRegistryCredential, password: '' },
+				continueOnFail: true,
+			}),
+		);
+
+		expect(result).toEqual([
+			[
+				expect.objectContaining({
+					json: { error: 'Username and password are required for Schema Registry Basic Auth' },
+				}),
+			],
+		]);
+	});
+});

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

@@ -0,0 +1,113 @@
+import type { KafkaJS } from '@confluentinc/kafka-javascript';
+
+import type { KafkaCredentials } from '../../../utils';
+import { getKafkaLibrary } from '../../../v2/transport/client';
+import { createKafkaProducer } from '../../../v2/transport/producer';
+import { confluentKafkaModuleMock } from '../../mocks/confluent-kafka';
+
+vi.mock('@confluentinc/kafka-javascript', () => confluentKafkaModuleMock());
+
+const CA_PEM = '-----BEGIN CERTIFICATE-----\nMIIBcacertbody==\n-----END CERTIFICATE-----';
+
+const credentials: KafkaCredentials = {
+	clientId: 'test',
+	brokers: 'localhost:9092',
+	ssl: false,
+	authentication: false,
+};
+
+// client.ts caches the library after the first call in this file, so re-reading it
+// here always returns the same `Kafka` mock constructor createKafkaProducer used.
+async function kafkaConstructorMock() {
+	const { Kafka } = await getKafkaLibrary();
+	return vi.mocked(Kafka);
+}
+
+async function lastKafkaInstance() {
+	return (await kafkaConstructorMock()).mock.results.at(-1)?.value;
+}
+
+describe('createKafkaProducer', () => {
+	beforeEach(() => {
+		vi.clearAllMocks();
+	});
+
+	it.each([
+		{ acks: 1, timeout: 30000 },
+		{ acks: 0, timeout: 5000 },
+	])('passes acks and timeout through under the kafkaJS key (%o)', async (options) => {
+		await createKafkaProducer(credentials, options);
+
+		const kafkaInstance = await lastKafkaInstance();
+		expect(kafkaInstance.producer).toHaveBeenCalledWith({
+			kafkaJS: { ...options, allowAutoTopicCreation: true },
+		});
+	});
+
+	it('applies the optional compression codec to the producer config', async () => {
+		await createKafkaProducer(credentials, {
+			acks: 1,
+			timeout: 30000,
+			compression: 'gzip' as KafkaJS.CompressionTypes,
+		});
+
+		const kafkaInstance = await lastKafkaInstance();
+		expect(kafkaInstance.producer).toHaveBeenCalledWith({
+			kafkaJS: { acks: 1, timeout: 30000, allowAutoTopicCreation: true, compression: 'gzip' },
+		});
+	});
+
+	it('builds the client from the credential', async () => {
+		await createKafkaProducer(credentials, { acks: 1, timeout: 30000 });
+
+		expect(await kafkaConstructorMock()).toHaveBeenCalledWith(
+			expect.objectContaining({
+				kafkaJS: expect.objectContaining({ brokers: ['localhost:9092'], clientId: 'test' }),
+			}),
+		);
+	});
+
+	it('hands the sasl and TLS material of the credential to the client', async () => {
+		await createKafkaProducer(
+			{
+				...credentials,
+				ssl: true,
+				ca: CA_PEM,
+				authentication: true,
+				saslMechanism: 'scram-sha-512',
+				username: 'user',
+				password: 'pass',
+			},
+			{ acks: 1, timeout: 30000 },
+		);
+
+		expect(await kafkaConstructorMock()).toHaveBeenCalledWith(
+			expect.objectContaining({
+				kafkaJS: expect.objectContaining({
+					ssl: true,
+					sasl: { mechanism: 'scram-sha-512', username: 'user', password: 'pass' },
+				}),
+				'ssl.ca.pem': CA_PEM,
+			}),
+		);
+	});
+
+	it('pins the library log level so it does not write broker details to stdout', async () => {
+		await createKafkaProducer(credentials, { acks: 1, timeout: 30000 });
+
+		const { logLevel } = await getKafkaLibrary();
+		expect(await kafkaConstructorMock()).toHaveBeenCalledWith(
+			expect.objectContaining({
+				kafkaJS: expect.objectContaining({ logLevel: logLevel.ERROR }),
+			}),
+		);
+	});
+
+	it('returns the client-built producer, unconnected', async () => {
+		const producer = await createKafkaProducer(credentials, { acks: 1, timeout: 30000 });
+
+		const kafkaInstance = await lastKafkaInstance();
+		expect(producer).toBe(kafkaInstance.producer.mock.results[0].value);
+		expect(producer.connect).not.toHaveBeenCalled();
+	});
+});

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

@@ -0,0 +1,436 @@
+import type { KafkaJS } from '@confluentinc/kafka-javascript';
+import type { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
+import { ensureError } from '@n8n/utils/errors/ensure-error';
+import type {
+	IExecuteFunctions,
+	IDataObject,
+	INode,
+	INodeExecutionData,
+	INodeType,
+	INodeTypeBaseDescription,
+	INodeTypeDescription,
+} from 'n8n-workflow';
+import { jsonParse, NodeConnectionTypes, NodeError, NodeOperationError } from 'n8n-workflow';
+
+import { generatePairedItemData } from '@utils/utilities';
+
+import { createSchemaRegistry, type KafkaCredentials } from '../utils';
+import { createKafkaProducer, type KafkaProducerOptions } from './transport';
+
+const DEFAULT_TIMEOUT_MS = 30000;
+
+/** One row of the `headersUi` fixed collection. */
+interface HeaderRow {
+	key: string;
+	value: string;
+}
+
+/**
+ * Maps the `options` collection onto the producer factory's options. Both fields
+ * differ deliberately from v1, so they are converted in one place rather than at
+ * the call site.
+ */
+function toProducerOptions(options: IDataObject): KafkaProducerOptions {
+	return {
+		// -1 = all in-sync replicas, matching the option description. v1 maps
+		// `true` to 1 (leader only) — a bug not worth carrying into a new version.
+		acks: options.acks === true ? -1 : 0,
+		// Unlike v1 (kafkajs tolerates `undefined`), confluent's native library
+		// crashes if `timeout` reaches the producer config as `undefined` — which
+		// it would be here if the user never added the option, since a `collection`
+		// param only carries keys the user explicitly set, ignoring its declared
+		// UI default. Fall back to that same default explicitly.
+		timeout: (options.timeout as number | undefined) ?? DEFAULT_TIMEOUT_MS,
+	};
+}
+
+/** Resolved Schema Registry client plus the schema id every message encodes against. */
+interface ResolvedSchemaRegistry {
+	registry: SchemaRegistry;
+	schemaId: number;
+}
+
+/**
+ * Encodes a message for the wire, returning it unchanged when the Schema Registry
+ * is off. Both failure modes are the user's to fix, so each maps to its own
+ * message rather than surfacing a registry-internal error.
+ */
+async function encodeMessage(
+	message: string,
+	schemaRegistry: ResolvedSchemaRegistry | undefined,
+	node: INode,
+	itemIndex: number,
+): Promise<string | Buffer> {
+	if (!schemaRegistry) return message;
+
+	let parsedMessage: unknown;
+	try {
+		parsedMessage = JSON.parse(message);
+	} catch {
+		throw new NodeOperationError(node, 'Message is not valid JSON', {
+			description:
+				'The Schema Registry encodes JSON messages. Provide a valid JSON message, or turn off "Use Schema Registry".',
+			itemIndex,
+		});
+	}
+
+	try {
+		return await schemaRegistry.registry.encode(schemaRegistry.schemaId, parsedMessage);
+	} catch {
+		// The original error is dropped rather than kept as `cause`: registry errors
+		// interpolate the request URL and response body, which would then be
+		// persisted into execution data.
+		throw new NodeOperationError(node, 'Verify your Schema Registry configuration', { itemIndex });
+	}
+}
+
+/**
+ * The native binding rejects a non-string header value only after the message is queued,
+ * so an unchecked value fails the node for a message the broker already accepted.
+ */
+function validateHeaders(headers: KafkaJS.IHeaders, node: INode, itemIndex: number) {
+	for (const [key, value] of Object.entries(headers)) {
+		const values = Array.isArray(value) ? value : [value];
+
+		if (values.some((entry) => typeof entry !== 'string' && !Buffer.isBuffer(entry))) {
+			throw new NodeOperationError(node, `Header "${key}" must be a string`, { itemIndex });
+		}
+	}
+}
+
+const versionDescription: INodeTypeDescription = {
+	displayName: 'Kafka',
+	name: 'kafka',
+	icon: { light: 'file:kafka.svg', dark: 'file:kafka.dark.svg' },
+	group: ['transform'],
+	version: 2,
+	description: 'Sends messages to a Kafka topic',
+	defaults: {
+		name: 'Kafka',
+	},
+	usableAsTool: true,
+	inputs: [NodeConnectionTypes.Main],
+	outputs: [NodeConnectionTypes.Main],
+	credentials: [
+		{
+			// Leave the `kafka` credential test to v1: it is resolved per credential type, so a
+			// `methods.credentialTest.kafkaConnectionTest` here would take over v1's test too.
+			// See 'should leave the kafka credential test to v1' in test/Kafka.node.test.ts.
+			name: 'kafka',
+			required: true,
+		},
+		{
+			name: 'schemaRegistryApi',
+			required: false,
+			displayName: 'Schema Registry',
+			displayOptions: {
+				show: {
+					useSchemaRegistry: [true],
+				},
+			},
+		},
+	],
+	properties: [
+		{
+			displayName: 'Topic',
+			name: 'topic',
+			type: 'string',
+			default: '',
+			placeholder: 'topic-name',
+			description: 'Name of the queue of topic to publish to',
+		},
+		{
+			displayName: 'Send Input Data',
+			name: 'sendInputData',
+			type: 'boolean',
+			default: true,
+			description: 'Whether to send the data the node receives as JSON to Kafka',
+		},
+		{
+			displayName: 'Message',
+			name: 'message',
+			type: 'string',
+			displayOptions: {
+				show: {
+					sendInputData: [false],
+				},
+			},
+			default: '',
+			description: 'The message to be sent',
+		},
+		{
+			displayName: 'JSON Parameters',
+			name: 'jsonParameters',
+			type: 'boolean',
+			default: false,
+		},
+		{
+			displayName: 'Use Schema Registry',
+			name: 'useSchemaRegistry',
+			type: 'boolean',
+			default: false,
+			description: 'Whether to use Confluent Schema Registry',
+		},
+		{
+			displayName: 'Schema Registry URL',
+			name: 'schemaRegistryUrl',
+			type: 'string',
+			displayOptions: {
+				show: {
+					useSchemaRegistry: [true],
+				},
+			},
+			placeholder: 'https://schema-registry-domain:8081',
+			default: '',
+			description:
+				'URL of the schema registry. Only used when no Schema Registry credential is selected.',
+		},
+		{
+			displayName: 'Use Key',
+			name: 'useKey',
+			type: 'boolean',
+			default: false,
+			description: 'Whether to use a message key',
+		},
+		{
+			displayName: 'Key',
+			name: 'key',
+			type: 'string',
+			required: true,
+			displayOptions: {
+				show: {
+					useKey: [true],
+				},
+			},
+			placeholder: '',
+			default: '',
+			description: 'The message key',
+		},
+		{
+			displayName: 'Event Name',
+			name: 'eventName',
+			type: 'string',
+			required: true,
+			displayOptions: {
+				show: {
+					useSchemaRegistry: [true],
+				},
+			},
+			default: '',
+			description: 'Namespace and Name of Schema in Schema Registry (namespace.name)',
+		},
+		{
+			displayName: 'Headers',
+			name: 'headersUi',
+			placeholder: 'Add Header',
+			type: 'fixedCollection',
+			displayOptions: {
+				show: {
+					jsonParameters: [false],
+				},
+			},
+			typeOptions: {
+				multipleValues: true,
+			},
+			default: {},
+			options: [
+				{
+					name: 'headerValues',
+					displayName: 'Header',
+					values: [
+						{
+							displayName: 'Key',
+							name: 'key',
+							type: 'string',
+							default: '',
+						},
+						{
+							displayName: 'Value',
+							name: 'value',
+							type: 'string',
+							default: '',
+						},
+					],
+				},
+			],
+		},
+		{
+			displayName: 'Headers (JSON)',
+			name: 'headerParametersJson',
+			type: 'json',
+			displayOptions: {
+				show: {
+					jsonParameters: [true],
+				},
+			},
+			default: '',
+			description: 'Header parameters as JSON (flat object)',
+		},
+		{
+			displayName: 'Options',
+			name: 'options',
+			type: 'collection',
+			default: {},
+			placeholder: 'Add option',
+			options: [
+				{
+					displayName: 'Acks',
+					name: 'acks',
+					type: 'boolean',
+					default: false,
+					description: 'Whether or not producer must wait for acknowledgement from all replicas',
+				},
+				{
+					displayName: 'Timeout',
+					name: 'timeout',
+					type: 'number',
+					default: DEFAULT_TIMEOUT_MS,
+					description: 'The time to await a response in ms',
+				},
+			],
+		},
+	],
+};
+
+export class KafkaV2 implements INodeType {
+	description: INodeTypeDescription;
+
+	constructor(baseDescription: INodeTypeBaseDescription) {
+		this.description = {
+			...baseDescription,
+			...versionDescription,
+		};
+	}
+
+	async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
+		const items = this.getInputData();
+		const itemData = generatePairedItemData(items.length);
+
+		const length = items.length;
+
+		const topicMessages: KafkaJS.TopicMessages[] = [];
+
+		let responseData: IDataObject[];
+
+		try {
+			const producerOptions = toProducerOptions(this.getNodeParameter('options', 0));
+			const sendInputData = this.getNodeParameter('sendInputData', 0) as boolean;
+
+			const useSchemaRegistry = this.getNodeParameter('useSchemaRegistry', 0) as boolean;
+
+			const credentials = await this.getCredentials<KafkaCredentials>('kafka');
+
+			// Resolve the registry configuration once, before the producer is set
+			// up, so credential misconfiguration surfaces with its own error
+			// message and never leaks a connected producer. The registry client
+			// and schema ID are loop-invariant (`eventName` is read at index 0)
+			let schemaRegistry: ResolvedSchemaRegistry | undefined;
+
+			if (useSchemaRegistry) {
+				const registry = await createSchemaRegistry(
+					this,
+					this.getNodeParameter('schemaRegistryUrl', 0) as string,
+				);
+
+				try {
+					const eventName = this.getNodeParameter('eventName', 0) as string;
+					const schemaId = await registry.getLatestSchemaId(eventName);
+					schemaRegistry = { registry, schemaId };
+				} catch (exception) {
+					throw new NodeOperationError(this.getNode(), 'Verify your Schema Registry configuration');
+				}
+			}
+
+			for (let i = 0; i < length; i++) {
+				const rawMessage = sendInputData
+					? JSON.stringify(items[i].json)
+					: (this.getNodeParameter('message', i) as string);
+
+				const message = await encodeMessage(rawMessage, schemaRegistry, this.getNode(), i);
+
+				const topic = this.getNodeParameter('topic', i) as string;
+
+				const jsonParameters = this.getNodeParameter('jsonParameters', i);
+
+				const useKey = this.getNodeParameter('useKey', i) as boolean;
+
+				const key = useKey ? (this.getNodeParameter('key', i) as string) : null;
+
+				let headers: KafkaJS.IHeaders;
+
+				if (jsonParameters) {
+					try {
+						headers = jsonParse<KafkaJS.IHeaders>(
+							this.getNodeParameter('headerParametersJson', i) as string,
+						);
+					} catch {
+						throw new NodeOperationError(this.getNode(), 'Headers must be valid JSON', {
+							itemIndex: i,
+						});
+					}
+				} else {
+					// `Object.fromEntries` builds the object in one step rather than assigning
+					// user-supplied names as computed keys.
+					headers = Object.fromEntries(
+						(
+							((this.getNodeParameter('headersUi', i) as IDataObject).headerValues ??
+								[]) as HeaderRow[]
+						).map(({ key: headerKey, value }) => [headerKey, value]),
+					);
+				}
+
+				validateHeaders(headers, this.getNode(), i);
+
+				topicMessages.push({
+					topic,
+					messages: [
+						{
+							value: message,
+							headers,
+							key,
+						},
+					],
+				});
+			}
+
+			const producer = await createKafkaProducer(credentials, producerOptions);
+
+			try {
+				await producer.connect();
+				responseData = await producer.sendBatch({ topicMessages });
+			} finally {
+				// Unlike v1, always close the connection. The failure is logged rather than
+				// rethrown so it can never mask the error the user needs to see — but a native
+				// client that fails to disconnect leaks threads, so it must leave a trace.
+				await producer.disconnect().catch((disconnectError) => {
+					this.logger.warn('Kafka producer failed to disconnect', {
+						error: ensureError(disconnectError).message,
+					});
+				});
+			}
+
+			if (responseData.length === 0) {
+				responseData.push({
+					success: true,
+				});
+			}
+
+			const executionData = this.helpers.constructExecutionMetaData(
+				this.helpers.returnJsonArray(responseData),
+				{ itemData },
+			);
+
+			return [executionData];
+		} catch (error) {
+			if (this.continueOnFail()) {
+				return [[{ json: { error: ensureError(error).message }, pairedItem: itemData }]];
+			}
+
+			// The transport throws plain UserErrors for an unusable credential, and core adds no
+			// node context to non-NodeErrors, so they would surface in the UI unattributed.
+			if (error instanceof NodeError) throw error;
+
+			throw new NodeOperationError(this.getNode(), ensureError(error));
+		}
+	}
+}

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

@@ -1,2 +1 @@
-export { toKafkaJSConfig } from './config';
-export { getKafkaLibrary } from './client';
+export { createKafkaProducer, type KafkaProducerOptions } from './producer';

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

@@ -0,0 +1,34 @@
+import type { KafkaJS } from '@confluentinc/kafka-javascript';
+
+import { getKafkaLibrary } from './client';
+import { toKafkaJSConfig } from './config';
+import type { KafkaCredentials } from '../../utils';
+
+export interface KafkaProducerOptions {
+	acks: number;
+	timeout: number;
+	// Must be set at construction: the library locks compression in when the producer is created.
+	compression?: KafkaJS.CompressionTypes;
+}
+
+export async function createKafkaProducer(
+	credentials: KafkaCredentials,
+	options: KafkaProducerOptions,
+): Promise<KafkaJS.Producer> {
+	const { Kafka, logLevel } = await getKafkaLibrary();
+	const config = toKafkaJSConfig(credentials);
+	// Without an explicit level the library's own logger writes broker host:port to
+	// process stdout on every execution, outside n8n's logger.
+	const kafka = new Kafka({ ...config, kafkaJS: { ...config.kafkaJS, logLevel: logLevel.ERROR } });
+
+	// acks and timeout are locked in at construction: the library's KafkaJS
+	// compatibility layer ignores them when passed to sendBatch.
+	return kafka.producer({
+		kafkaJS: {
+			acks: options.acks,
+			timeout: options.timeout,
+			allowAutoTopicCreation: true,
+			...(options.compression && { compression: options.compression }),
+		},
+	});
+}