feat(Kafka Trigger Node): Add v2 node class and register version 2 (no-changelog) (#35715)

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

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

@@ -2,6 +2,7 @@ import type { INodeTypeBaseDescription, IVersionedNodeType } from 'n8n-workflow'
 import { VersionedNodeType } from 'n8n-workflow';
 
 import { KafkaTriggerV1 } from './v1/KafkaTriggerV1.node';
+import { KafkaTriggerV2 } from './v2/KafkaTriggerV2.node';
 
 export class KafkaTrigger extends VersionedNodeType {
 	constructor() {
@@ -19,6 +20,7 @@ export class KafkaTrigger extends VersionedNodeType {
 			1.1: new KafkaTriggerV1(baseDescription),
 			1.2: new KafkaTriggerV1(baseDescription),
 			1.3: new KafkaTriggerV1(baseDescription),
+			2: new KafkaTriggerV2(baseDescription),
 		};
 
 		super(nodeVersions, baseDescription);

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

@@ -24,6 +24,7 @@ import { testTriggerNode } from '@test/nodes/TriggerHelpers';
 
 import { KafkaTrigger } from '../KafkaTrigger.node';
 import { KafkaTriggerV1 } from '../v1/KafkaTriggerV1.node';
+import { KafkaTriggerV2 } from '../v2/KafkaTriggerV2.node';
 import type { Mock, Mocked } from 'vitest';
 
 vi.mock('kafkajs');
@@ -2528,24 +2529,31 @@ describe('KafkaTrigger Node', () => {
 
 describe('KafkaTrigger (versioned entry point)', () => {
 	const kafkaTrigger = new KafkaTrigger();
-	const expectedDescription = new KafkaTriggerV1(baseDescription).description;
-	const versions = [1, 1.1, 1.2, 1.3];
+	const expectedV1Description = new KafkaTriggerV1(baseDescription).description;
+	const expectedV2Description = new KafkaTriggerV2(baseDescription).description;
+	const v1Versions = [1, 1.1, 1.2, 1.3];
+	const allVersions = [...v1Versions, 2];
 
-	it('maps exactly versions 1, 1.1, 1.2, and 1.3 to KafkaTriggerV1', () => {
+	it('maps exactly versions 1, 1.1, 1.2, 1.3 to KafkaTriggerV1 and 2 to KafkaTriggerV2', () => {
 		expect(
 			Object.keys(kafkaTrigger.nodeVersions)
 				.map(Number)
 				.sort((a, b) => a - b),
-		).toEqual(versions);
-		for (const version of versions) {
+		).toEqual(allVersions);
+		for (const version of v1Versions) {
 			expect(kafkaTrigger.nodeVersions[version]).toBeInstanceOf(KafkaTriggerV1);
 		}
+		expect(kafkaTrigger.nodeVersions[2]).toBeInstanceOf(KafkaTriggerV2);
+	});
+
+	it('resolves each v1.x version to a consistent, correctly-merged description', () => {
+		for (const version of v1Versions) {
+			expect(kafkaTrigger.nodeVersions[version].description).toEqual(expectedV1Description);
+		}
 	});
 
-	it('resolves each version to a consistent, correctly-merged description', () => {
-		for (const version of versions) {
-			expect(kafkaTrigger.nodeVersions[version].description).toEqual(expectedDescription);
-		}
+	it('resolves version 2 to a consistent, correctly-merged description', () => {
+		expect(kafkaTrigger.nodeVersions[2].description).toEqual(expectedV2Description);
 	});
 
 	it('defaults new workflows to version 1.3', () => {

packages/nodes-base/nodes/Kafka/test/v2/KafkaTriggerIsolation.test.ts

@@ -0,0 +1,87 @@
+// Static import of the entry file, not a version file directly: constructing
+// `KafkaTrigger` builds every registered version's class (KafkaTriggerV1 x4
+// AND KafkaTriggerV2), which is the scenario that must not touch the new
+// library. The node is imported directly (through vite), not via
+// NodeTestHarness (which loads from dist via require()), so vi.mock can
+// intercept its imports - same reasoning as test/v2/KafkaV2.node.test.ts.
+import { testTriggerNode } from '@test/nodes/TriggerHelpers';
+
+import { KafkaTrigger } from '../../KafkaTrigger.node';
+import {
+	confluentKafkaModuleMock,
+	getConfluentKafkaAccessCount,
+	resetConfluentKafkaAccessCount,
+} from '../mocks/confluent-kafka';
+
+vi.mock('@confluentinc/kafka-javascript', () => confluentKafkaModuleMock());
+
+// A minimal but complete kafkajs consumer: v1's trigger() registers event
+// listeners via consumer.on/consumer.events, so a bare vi.mock('kafkajs')
+// automock (whose methods return undefined) fails before the loop even
+// starts. This test only needs v1 to activate successfully, not to receive
+// a message.
+vi.mock('kafkajs', () => {
+	const events = {
+		CONNECT: 'consumer.connect',
+		GROUP_JOIN: 'consumer.group_join',
+		REQUEST_TIMEOUT: 'consumer.network.request_timeout',
+		RECEIVED_UNSUBSCRIBED_TOPICS: 'consumer.received_unsubscribed_topics',
+		STOP: 'consumer.stop',
+		DISCONNECT: 'consumer.disconnect',
+		COMMIT_OFFSETS: 'consumer.commit_offsets',
+		REBALANCING: 'consumer.rebalancing',
+		CRASH: 'consumer.crash',
+	};
+	const consumer = {
+		connect: vi.fn(async () => {}),
+		subscribe: vi.fn(async () => {}),
+		run: vi.fn(async () => {}),
+		stop: vi.fn(async () => {}),
+		disconnect: vi.fn(async () => {}),
+		on: vi.fn(() => vi.fn()),
+		events,
+	};
+	// A function expression, not an arrow: v1 calls `new Kafka(...)`, and an
+	// arrow implementation is not constructible.
+	return {
+		Kafka: vi.fn(function () {
+			return { consumer: vi.fn(() => consumer) };
+		}),
+		logLevel: { NOTHING: 0, ERROR: 1, WARN: 2, INFO: 3, DEBUG: 4 },
+	};
+});
+
+beforeEach(() => {
+	resetConfluentKafkaAccessCount();
+});
+
+it('running a version 1 trigger never loads the new confluent-kafka library', async () => {
+	const entry = new KafkaTrigger();
+	const v1 = entry.nodeVersions[1];
+
+	expect(getConfluentKafkaAccessCount()).toBe(0);
+
+	const { close } = await testTriggerNode(v1, {
+		mode: 'trigger',
+		node: {
+			typeVersion: 1,
+			parameters: {
+				topic: 'isolation-topic',
+				groupId: 'isolation-test-group-v1',
+				useSchemaRegistry: false,
+			},
+		},
+		credential: {
+			brokers: 'localhost:9092',
+			clientId: 'n8n-isolation-test',
+			ssl: false,
+			authentication: false,
+		},
+	});
+
+	expect(getConfluentKafkaAccessCount()).toBe(0);
+
+	await close();
+
+	expect(getConfluentKafkaAccessCount()).toBe(0);
+});

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

@@ -0,0 +1,99 @@
+import type {
+	INodeParameters,
+	INodeProperties,
+	INodeTypeBaseDescription,
+	INodeTypeDescription,
+} from 'n8n-workflow';
+import { NodeHelpers } from 'n8n-workflow';
+
+import { KafkaTriggerV1 } from '../../v1/KafkaTriggerV1.node';
+import { KafkaTriggerV2 } from '../../v2/KafkaTriggerV2.node';
+
+const baseDescription: INodeTypeBaseDescription = {
+	displayName: 'Kafka Trigger',
+	name: 'kafkaTrigger',
+	icon: { light: 'file:kafka.svg', dark: 'file:kafka.dark.svg' },
+	group: ['trigger'],
+	defaultVersion: 1.3,
+	description: 'Consume messages from a Kafka topic',
+};
+
+/**
+ * v1.3 options v2 does not carry, because none of them could do anything here.
+ * Auto Commit Threshold has no equivalent in the new library. Each Batch Auto
+ * Resolve cannot be honoured: the consume loop resolves offsets chunk by chunk
+ * and turns the library's automatic resolution off, so obeying it would mark
+ * messages read that no execution ever saw. Allow Topic Creation reaches
+ * librdkafka but changed nothing when measured against a real broker, with the
+ * flag on or off.
+ */
+const DROPPED_IN_V2 = ['autoCommitThreshold', 'eachBatchAutoResolve', 'allowAutoTopicCreation'];
+
+/** v2 adds no options of its own; it only drops the three above. */
+const ADDED_IN_V2: string[] = [];
+
+/** The entries of the `options` collection on a resolved node description. */
+function optionEntries(description: INodeTypeDescription): INodeProperties[] {
+	const collection = description.properties.find((property) => property.name === 'options');
+	if (!collection?.options) throw new Error('the description declares no Options collection');
+	return collection.options as INodeProperties[];
+}
+
+/**
+ * The option names a user actually sees at a given `typeVersion`.
+ *
+ * v1's raw array holds every minor version's fields at once, including two
+ * entries both named `heartbeatInterval` and a `parallelProcessing` restricted
+ * to 1.1 and 1.2, so it cannot be compared directly. `displayParameter` is what
+ * n8n itself uses to decide visibility, so the version predicates are resolved
+ * the same way here rather than being restated by hand.
+ *
+ * The sibling values below are chosen so every predicate that depends on another
+ * field rather than on the version resolves to visible: Only Message needs JSON
+ * Parse Message on, and Retry Delay on Error is hidden only for the
+ * `immediately` offset mode. This is the full set a user could see, so nothing
+ * is missed for depending on a sibling.
+ */
+function visibleOptionNames(description: INodeTypeDescription, typeVersion: number): string[] {
+	const values: INodeParameters = { jsonParseMessage: true };
+	const root: INodeParameters = { ...values, resolveOffset: 'onCompletion' };
+
+	return optionEntries(description)
+		.filter((option) =>
+			NodeHelpers.displayParameter(values, option, { typeVersion }, description, root),
+		)
+		.map((option) => option.name);
+}
+
+describe('KafkaTriggerV2 description', () => {
+	const v13 = new KafkaTriggerV1(baseDescription).description;
+	const v2 = new KafkaTriggerV2(baseDescription).description;
+
+	it('carries every v1.3 option except the three that control nothing', () => {
+		const expected = visibleOptionNames(v13, 1.3).filter((name) => !DROPPED_IN_V2.includes(name));
+		const actual = visibleOptionNames(v2, 2).filter((name) => !ADDED_IN_V2.includes(name));
+
+		expect(actual).toStrictEqual(expected);
+	});
+
+	it('differs from v1.3 by exactly the documented options, and nothing else', () => {
+		const v13Visible = visibleOptionNames(v13, 1.3);
+		const v2Visible = visibleOptionNames(v2, 2);
+
+		// Guards the helper itself: if the version resolution ever silently returned
+		// nothing, the assertion above would pass against two empty lists.
+		expect(v13Visible.length).toBeGreaterThan(10);
+		expect(v13Visible).toEqual(expect.arrayContaining(DROPPED_IN_V2));
+		// Sorted: this is about which names differ, not declaration order.
+		expect(v13Visible.filter((name) => !v2Visible.includes(name)).sort()).toStrictEqual(
+			[...DROPPED_IN_V2].sort(),
+		);
+		expect(v2Visible.filter((name) => !v13Visible.includes(name)).sort()).toStrictEqual(
+			[...ADDED_IN_V2].sort(),
+		);
+	});
+
+	it('never shows Parallel Processing, which v1 restricts to 1.1 and 1.2', () => {
+		expect(visibleOptionNames(v2, 2)).not.toContain('parallelProcessing');
+	});
+});

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

@@ -0,0 +1,994 @@
+import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
+import type { INodeTypeBaseDescription, IRun, Logger } from 'n8n-workflow';
+import { TriggerCloseError, UserError } from 'n8n-workflow';
+import type { Mock, Mocked } from 'vitest';
+import { mock } from 'vitest-mock-extended';
+
+import { testTriggerNode } from '@test/nodes/TriggerHelpers';
+
+import { DEFAULT_EXECUTION_TIMEOUT_SECONDS } from '../../v2/consumer';
+import { KafkaTriggerV2 } from '../../v2/KafkaTriggerV2.node';
+import {
+	explainManualRunGroupDenial,
+	manualRunGroupId,
+	toConsumerOptions,
+	toEmitterOptions,
+} from '../../v2/TriggerSettings';
+import {
+	confluentKafkaModuleMock,
+	getFakeConsumers,
+	resetConfluentKafkaRecordings,
+	type FakeConsumer,
+} from '../mocks/confluent-kafka';
+
+vi.mock('@confluentinc/kafka-javascript', () => confluentKafkaModuleMock());
+vi.mock('@kafkajs/confluent-schema-registry');
+
+// Wraps the real consumeTopic in a spy rather than replacing it, so the actual
+// loop still drives these tests. Only used for options that reach the loop but
+// leave no trace on the fake consumer, such as errorRetryDelay.
+const { consumeTopicSpy } = vi.hoisted(() => ({ consumeTopicSpy: vi.fn() }));
+vi.mock('../../v2/consumer', async (importOriginal) => {
+	const actual = await importOriginal<typeof import('../../v2/consumer')>();
+	return {
+		...actual,
+		consumeTopic: vi.fn(async (...args: Parameters<typeof actual.consumeTopic>) => {
+			consumeTopicSpy(...args);
+			return await actual.consumeTopic(...args);
+		}),
+	};
+});
+
+const baseDescription: INodeTypeBaseDescription = {
+	displayName: 'Kafka Trigger',
+	name: 'kafkaTrigger',
+	icon: { light: 'file:kafka.svg', dark: 'file:kafka.dark.svg' },
+	group: ['trigger'],
+	defaultVersion: 1.3,
+	description: 'Consume messages from a Kafka topic',
+};
+
+const credential = {
+	brokers: 'localhost:9092',
+	clientId: 'n8n-kafka',
+	ssl: false,
+	authentication: false,
+};
+
+async function lastFakeConsumer(): Promise<FakeConsumer> {
+	const consumer = getFakeConsumers().at(-1);
+	if (!consumer) throw new Error('the fake recorded no consumer');
+	return consumer;
+}
+
+/**
+ * Starts the trigger. Defaults to `immediately`, so the emitter does not wait on
+ * an execution: tests about parsing and consumer settings can then deliver a
+ * batch without also having to resolve a run.
+ */
+async function startTrigger(
+	groupId: string,
+	parameters: Record<string, unknown> = {},
+	overrides: Parameters<typeof testTriggerNode>[1] = {},
+) {
+	return await testTriggerNode(new KafkaTriggerV2(baseDescription), {
+		mode: 'trigger',
+		node: {
+			parameters: {
+				topic: 'test-topic',
+				groupId,
+				useSchemaRegistry: false,
+				resolveOffset: 'immediately',
+				...parameters,
+			},
+		},
+		credential,
+		...overrides,
+	});
+}
+
+describe('toConsumerOptions', () => {
+	it("applies v1's consumer defaults when the user set nothing", () => {
+		const result = toConsumerOptions({}, 'my-group', undefined);
+
+		expect(result).toStrictEqual({
+			groupId: 'my-group',
+			sessionTimeout: 30000,
+			// v1.3's default, not the 3000 v1 uses below 1.3
+			heartbeatInterval: 10000,
+			// No workflow timeout and no option, so the emitter's own default wait
+			// stands in, halved because the library doubles it
+			rebalanceTimeout: 1_800_000,
+			maxBytesPerPartition: undefined,
+			minBytes: undefined,
+			maxInFlightRequests: undefined,
+			fromBeginning: undefined,
+			autoCommitInterval: undefined,
+		});
+	});
+
+	it('passes the user-set consumer options through', () => {
+		const result = toConsumerOptions(
+			{
+				sessionTimeout: 20000,
+				heartbeatInterval: 2000,
+				fetchMaxBytes: 2097152,
+				fetchMinBytes: 1024,
+				maxInFlightRequests: 5,
+				fromBeginning: true,
+			},
+			'my-group',
+			undefined,
+		);
+
+		expect(result).toMatchObject({
+			sessionTimeout: 20000,
+			heartbeatInterval: 2000,
+			maxBytesPerPartition: 2097152,
+			minBytes: 1024,
+			maxInFlightRequests: 5,
+			fromBeginning: true,
+		});
+	});
+
+	it('halves the workflow execution timeout, since the library doubles it', () => {
+		// 600s of workflow timeout must stay 600s of processing headroom, and the
+		// library sets max.poll.interval.ms to twice whatever it is handed.
+		const result = toConsumerOptions({}, 'my-group', 600);
+
+		expect(result.rebalanceTimeout).toBe(300000);
+	});
+
+	it('falls back to the Rebalance Timeout option when the workflow timeout is unbounded', () => {
+		// n8n treats <= 0 as explicitly unbounded, and there is no deadline to derive
+		// from, so the node's own option decides.
+		const result = toConsumerOptions({ rebalanceTimeout: 900000 }, 'my-group', -1);
+
+		expect(result.rebalanceTimeout).toBe(450000);
+	});
+
+	describe('the processing deadline stays inside what the library accepts', () => {
+		// librdkafka takes 1..86400000 for max.poll.interval.ms and the library
+		// doubles what it is given, so anything past 12 hours here overflows the
+		// 32-bit int it is stored in. Measured against a real broker with a 30 day
+		// workflow timeout: "value -1702967296 is outside allowed range 1..86400000",
+		// and the consumer refuses to connect.
+		const LIBRDKAFKA_MAX_MS = 86_400_000;
+
+		it.each([
+			['30 days', 30 * 24 * 3600],
+			['a year', 365 * 24 * 3600],
+		])('caps a %s workflow timeout rather than overflowing', (_label, seconds) => {
+			const logger = mock<Logger>();
+
+			const { rebalanceTimeout } = toConsumerOptions({}, 'my-group', seconds, logger);
+
+			expect(rebalanceTimeout).toBe(43_200_000);
+			// What the library will actually hand librdkafka, after doubling. The
+			// `| 0` is the 32-bit truncation that turned the old value negative.
+			const doubled = (rebalanceTimeout ?? 0) * 2;
+			expect(doubled).toBeLessThanOrEqual(LIBRDKAFKA_MAX_MS);
+			expect(doubled | 0).toBeGreaterThan(0);
+			expect(logger.warn).toHaveBeenCalledWith(
+				expect.stringContaining('capped'),
+				expect.objectContaining({ appliedMs: LIBRDKAFKA_MAX_MS }),
+			);
+		});
+
+		it('caps an oversized Rebalance Timeout option too', () => {
+			const { rebalanceTimeout } = toConsumerOptions(
+				{ rebalanceTimeout: 30 * 24 * 3600 * 1000 },
+				'my-group',
+				undefined,
+			);
+
+			expect((rebalanceTimeout ?? 0) * 2).toBeLessThanOrEqual(LIBRDKAFKA_MAX_MS);
+		});
+
+		it('leaves a deadline inside the range alone, and says nothing', () => {
+			const logger = mock<Logger>();
+
+			const { rebalanceTimeout } = toConsumerOptions({}, 'my-group', 600, logger);
+
+			expect(rebalanceTimeout).toBe(300_000);
+			expect(logger.warn).not.toHaveBeenCalled();
+		});
+
+		it.each([NaN, Infinity])(
+			'falls back to the emitter default when Rebalance Timeout is %s',
+			(rebalanceTimeout) => {
+				// It can come from an expression, so it is not trusted to be usable.
+				const result = toConsumerOptions({ rebalanceTimeout }, 'my-group', undefined);
+
+				expect(result.rebalanceTimeout).toBe(1_800_000);
+			},
+		);
+
+		it('gives the broker the same deadline the emitter is prepared to wait', () => {
+			// These used to disagree: the broker got the Rebalance Timeout default of 10
+			// minutes while the emitter waited an hour, so an execution in between was
+			// fenced and its message redelivered while n8n believed the run owned it.
+			const { rebalanceTimeout } = toConsumerOptions({}, 'my-group', undefined);
+
+			// Doubled, because that is what the library hands librdkafka.
+			expect((rebalanceTimeout ?? 0) * 2).toBe(DEFAULT_EXECUTION_TIMEOUT_SECONDS * 1000);
+		});
+
+		describe('and stays above the Session Timeout, which the library also requires', () => {
+			// librdkafka refuses max.poll.interval.ms < session.timeout.ms on the classic
+			// group protocol (rdkafka_conf.c:4257). Both values are individually legal, so
+			// only the pair is wrong, and the consumer refuses to connect.
+			it.each([
+				['a 20s workflow timeout against the 30s session default', 20, undefined, 30_000],
+				['a 5s workflow timeout', 5, undefined, 30_000],
+			])('raises %s to the session timeout', (_label, seconds, session, expectedFloor) => {
+				const logger = mock<Logger>();
+
+				const { rebalanceTimeout } = toConsumerOptions(
+					session === undefined ? {} : { sessionTimeout: session },
+					'g',
+					seconds,
+					logger,
+				);
+
+				expect((rebalanceTimeout ?? 0) * 2).toBeGreaterThanOrEqual(expectedFloor);
+				expect(logger.warn).toHaveBeenCalledWith(
+					expect.stringContaining('raised to the Session Timeout'),
+					expect.objectContaining({ appliedMs: expectedFloor }),
+				);
+			});
+
+			it('raises a too-small Rebalance Timeout option the same way', () => {
+				const { rebalanceTimeout } = toConsumerOptions({ rebalanceTimeout: 2_000 }, 'g', undefined);
+
+				expect((rebalanceTimeout ?? 0) * 2).toBeGreaterThanOrEqual(30_000);
+			});
+
+			it('respects a lowered Session Timeout instead of forcing the 30s default', () => {
+				const logger = mock<Logger>();
+
+				const { rebalanceTimeout, sessionTimeout } = toConsumerOptions(
+					{ sessionTimeout: 8_000 },
+					'g',
+					5,
+					logger,
+				);
+
+				// 5s of workflow timeout is below an 8s session, so 8s is the floor.
+				expect((rebalanceTimeout ?? 0) * 2).toBeGreaterThanOrEqual(sessionTimeout ?? 0);
+				expect((rebalanceTimeout ?? 0) * 2).toBe(8_000);
+			});
+
+			it('says nothing when the deadline already clears the session timeout', () => {
+				const logger = mock<Logger>();
+
+				toConsumerOptions({}, 'g', 600, logger);
+
+				expect(logger.warn).not.toHaveBeenCalled();
+			});
+		});
+
+		it('lets an explicitly set Rebalance Timeout win when there is no workflow timeout', () => {
+			const { rebalanceTimeout } = toConsumerOptions(
+				{ rebalanceTimeout: 120_000 },
+				'my-group',
+				undefined,
+			);
+
+			expect(rebalanceTimeout).toBe(60_000);
+		});
+
+		it('falls back to the option when the workflow timeout is not a usable number', () => {
+			const result = toConsumerOptions({ rebalanceTimeout: 120_000 }, 'my-group', NaN);
+
+			expect(result.rebalanceTimeout).toBe(60_000);
+		});
+	});
+
+	describe('Auto Commit Interval', () => {
+		// v1 honours this too, via getAutoCommitSettings. What differs is the range
+		// check: librdkafka takes 0..86400000 and refuses the whole connection on
+		// anything else, naming a number the user never typed, so an unusable value
+		// is dropped with a warning here instead.
+		it('passes a user-set interval through, overriding the pinned default', () => {
+			expect(
+				toConsumerOptions({ autoCommitInterval: 1_000 }, 'g', undefined).autoCommitInterval,
+			).toBe(1_000);
+		});
+
+		it('keeps a zero, which turns interval commits off rather than meaning unset', () => {
+			// The loop still commits per chunk, so 0 is a real choice and must not be
+			// swallowed the way a falsy maxInFlightRequests is.
+			expect(toConsumerOptions({ autoCommitInterval: 0 }, 'g', undefined).autoCommitInterval).toBe(
+				0,
+			);
+		});
+
+		it('leaves the key off when the user set nothing, so the pinned default stands', () => {
+			expect(toConsumerOptions({}, 'g', undefined).autoCommitInterval).toBeUndefined();
+		});
+
+		it.each([NaN, -1, 86_400_001, Infinity])(
+			'drops %s, which the library would reject',
+			(value) => {
+				const logger = mock<Logger>();
+
+				const result = toConsumerOptions({ autoCommitInterval: value }, 'g', undefined, logger);
+
+				expect(result.autoCommitInterval).toBeUndefined();
+				expect(logger.warn).toHaveBeenCalledWith(
+					expect.stringContaining('Auto Commit Interval'),
+					expect.objectContaining({ supplied: value }),
+				);
+			},
+		);
+	});
+
+	describe('Heartbeat Interval stays under a third of the Session Timeout', () => {
+		// The two options are independent in the UI but not in Kafka, and pairing them
+		// badly fails silently: the broker fences the consumer, the offset is never
+		// committed, and the same message is redelivered forever. Measured against a
+		// real broker with a 10s session and the 10s heartbeat default, a 5s workflow
+		// re-ran one message every ~10s indefinitely.
+		it('leaves the defaults alone, which already sit at the recommended ratio', () => {
+			const result = toConsumerOptions({}, 'g', undefined);
+
+			expect(result).toMatchObject({ sessionTimeout: 30_000, heartbeatInterval: 10_000 });
+		});
+
+		it('lowers the default heartbeat when the user shortens only the session timeout', () => {
+			const logger = mock<Logger>();
+
+			const result = toConsumerOptions({ sessionTimeout: 10_000 }, 'g', undefined, logger);
+
+			expect(result.heartbeatInterval).toBe(3_333);
+			expect(logger.warn).toHaveBeenCalledWith(
+				expect.stringContaining('Heartbeat Interval'),
+				expect.objectContaining({ supplied: 10_000, applied: 3_333, sessionTimeout: 10_000 }),
+			);
+		});
+
+		it.each([
+			[30_000, 10_000, 10_000],
+			[30_000, 3_000, 3_000],
+			[9_000, 3_000, 3_000],
+		])('leaves %ims / %ims as set, since it is within the ratio', (session, beat, expected) => {
+			const logger = mock<Logger>();
+
+			const result = toConsumerOptions(
+				{ sessionTimeout: session, heartbeatInterval: beat },
+				'g',
+				undefined,
+				logger,
+			);
+
+			expect(result.heartbeatInterval).toBe(expected);
+			expect(logger.warn).not.toHaveBeenCalled();
+		});
+
+		it.each([
+			[10_000, 20_000, 3_333],
+			[10_000, 10_000, 3_333],
+			[6_000, 6_000, 2_000],
+		])('clamps %ims / %ims down to %ims', (session, beat, expected) => {
+			const result = toConsumerOptions(
+				{ sessionTimeout: session, heartbeatInterval: beat },
+				'g',
+				undefined,
+			);
+
+			expect(result.heartbeatInterval).toBe(expected);
+		});
+
+		it.each([NaN, Infinity])('clamps a %s heartbeat rather than forwarding it', (beat) => {
+			const result = toConsumerOptions({ heartbeatInterval: beat }, 'g', undefined);
+
+			expect(result.heartbeatInterval).toBe(10_000);
+		});
+
+		it.each([0, -1, NaN])(
+			'leaves the heartbeat alone when the session timeout is %s, so librdkafka names the bad value',
+			(session) => {
+				const logger = mock<Logger>();
+
+				const result = toConsumerOptions(
+					{ sessionTimeout: session, heartbeatInterval: 10_000 },
+					'g',
+					undefined,
+					logger,
+				);
+
+				expect(result).toMatchObject({ sessionTimeout: session, heartbeatInterval: 10_000 });
+				expect(logger.warn).not.toHaveBeenCalled();
+			},
+		);
+	});
+
+	it('drops a zero Max Number of Requests instead of forwarding it', () => {
+		// v1 turns 0 into null to mean "no limit". The library has no such sentinel,
+		// and a present-but-undefined key makes librdkafka fail.
+		const result = toConsumerOptions({ maxInFlightRequests: 0 }, 'my-group', undefined);
+
+		expect(result.maxInFlightRequests).toBeUndefined();
+	});
+});
+
+describe('manualRunGroupId', () => {
+	it('uses the configured group for an activated workflow', () => {
+		expect(manualRunGroupId('orders-consumer', false)).toBe('orders-consumer');
+	});
+
+	it('gives a manual run its own group, so it cannot take production offsets', () => {
+		const first = manualRunGroupId('orders-consumer', true);
+		const second = manualRunGroupId('orders-consumer', true);
+
+		expect(first).toMatch(/^orders-consumer-n8n-manual-.+/);
+		expect(first).not.toBe('orders-consumer');
+		// Two editors testing at once must not land in the same group either.
+		expect(second).not.toBe(first);
+	});
+});
+
+describe('explainManualRunGroupDenial', () => {
+	// A cluster with an authorizer usually grants group ACLs LITERAL on the exact
+	// Group ID, so the throwaway group is a resource nobody authorized. Only test
+	// runs break, and the broker's message names neither the group nor the fix.
+	const denial = new Error('Broker: Group authorization failed');
+
+	it('names the prefix to grant, so the fix does not need guessing', () => {
+		const result = explainManualRunGroupDenial(denial, 'orders-consumer', true);
+
+		expect(result).not.toBe(denial);
+		expect(result).toBeInstanceOf(UserError);
+		expect((result as UserError).description).toContain('orders-consumer-n8n-manual-');
+		// The original stays reachable rather than being replaced outright.
+		expect((result as UserError).cause).toBe(denial);
+	});
+
+	it('leaves an activated workflow alone, since its group is what the user typed', () => {
+		expect(explainManualRunGroupDenial(denial, 'orders-consumer', false)).toBe(denial);
+	});
+
+	it.each([
+		'Broker: Topic authorization failed',
+		'Broker: Not authorized to access cluster',
+		'SASL authentication failed',
+		'Broker: Unknown topic or partition',
+	])('leaves %s alone, since the group is not what was refused', (message) => {
+		const other = new Error(message);
+
+		expect(explainManualRunGroupDenial(other, 'orders-consumer', true)).toBe(other);
+	});
+});
+
+describe('toEmitterOptions', () => {
+	it('passes the execution timeout through raw, so unbounded stays unbounded', () => {
+		expect(toEmitterOptions({}, 'onCompletion', [], 0).executionTimeoutSeconds).toBe(0);
+		expect(
+			toEmitterOptions({}, 'onCompletion', [], undefined).executionTimeoutSeconds,
+		).toBeUndefined();
+		expect(toEmitterOptions({}, 'onCompletion', [], 120).executionTimeoutSeconds).toBe(120);
+	});
+
+	it('carries the mode, allowed statuses and retry delay', () => {
+		const result = toEmitterOptions({ errorRetryDelay: 1234 }, 'onStatus', ['success'], 60);
+
+		expect(result).toStrictEqual({
+			resolveOffsetMode: 'onStatus',
+			allowedStatuses: ['success'],
+			executionTimeoutSeconds: 60,
+			errorRetryDelay: 1234,
+		});
+	});
+});
+
+describe('KafkaTriggerV2 Node', () => {
+	beforeEach(() => {
+		resetConfluentKafkaRecordings();
+		consumeTopicSpy.mockClear();
+	});
+
+	it('connects, subscribes to the topic, and emits a received message', async () => {
+		const { emit, close } = await startTrigger('v2-basic');
+
+		const consumer = await lastFakeConsumer();
+		expect(consumer.connect).toHaveBeenCalledTimes(1);
+		expect(consumer.subscribe).toHaveBeenCalledWith({ topics: ['test-topic'] });
+		expect(consumer.run).toHaveBeenCalledTimes(1);
+
+		await consumer.deliverBatch({
+			topic: 'test-topic',
+			messages: [{ value: Buffer.from('message') }],
+		});
+
+		expect(emit).toHaveBeenCalledWith([[{ json: { message: 'message', topic: 'test-topic' } }]]);
+
+		await close();
+		expect(consumer.disconnect).toHaveBeenCalled();
+	});
+
+	describe('Batch Size', () => {
+		it('starts one execution per message by default, as v1 does', async () => {
+			const { emit } = await startTrigger('v2-batch-default');
+			const consumer = await lastFakeConsumer();
+
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [
+					{ value: Buffer.from('message1') },
+					{ value: Buffer.from('message2') },
+					{ value: Buffer.from('message3') },
+				],
+			});
+
+			// A 3-message library batch must not collapse into one 3-item execution.
+			expect(emit).toHaveBeenCalledTimes(3);
+			expect(emit).toHaveBeenNthCalledWith(1, [
+				[{ json: { message: 'message1', topic: 'test-topic' } }],
+			]);
+			expect(emit).toHaveBeenNthCalledWith(3, [
+				[{ json: { message: 'message3', topic: 'test-topic' } }],
+			]);
+		});
+
+		it('chunks into executions of Batch Size items when set above 1', async () => {
+			const { emit } = await startTrigger('v2-batch-2', { options: { batchSize: 2 } });
+			const consumer = await lastFakeConsumer();
+
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [
+					{ value: Buffer.from('message1') },
+					{ value: Buffer.from('message2') },
+					{ value: Buffer.from('message3') },
+				],
+			});
+
+			expect(emit).toHaveBeenCalledTimes(2);
+			expect(emit).toHaveBeenNthCalledWith(1, [
+				[
+					{ json: { message: 'message1', topic: 'test-topic' } },
+					{ json: { message: 'message2', topic: 'test-topic' } },
+				],
+			]);
+			expect(emit).toHaveBeenNthCalledWith(2, [
+				[{ json: { message: 'message3', topic: 'test-topic' } }],
+			]);
+		});
+	});
+
+	describe('consumer options reach the library', () => {
+		it('hands the user-set consumer options to the factory', async () => {
+			await startTrigger('v2-consumer-options', {
+				options: {
+					sessionTimeout: 20000,
+					heartbeatInterval: 2000,
+					fetchMaxBytes: 2097152,
+					fetchMinBytes: 1024,
+					maxInFlightRequests: 5,
+					fromBeginning: true,
+				},
+			});
+
+			const consumer = await lastFakeConsumer();
+			expect(consumer.config.kafkaJS).toMatchObject({
+				groupId: 'v2-consumer-options',
+				sessionTimeout: 20000,
+				heartbeatInterval: 2000,
+				maxBytesPerPartition: 2097152,
+				minBytes: 1024,
+				maxInFlightRequests: 5,
+				fromBeginning: true,
+			});
+		});
+
+		it('passes a caller-chosen partition concurrency to the loop', async () => {
+			await startTrigger('v2-concurrency', {
+				options: { partitionsConsumedConcurrently: 4 },
+			});
+
+			const consumer = await lastFakeConsumer();
+			expect(consumer.runConfig?.partitionsConsumedConcurrently).toBe(4);
+		});
+
+		it('passes Retry Delay on Error to the loop', async () => {
+			await startTrigger('v2-retry-delay', { options: { errorRetryDelay: 12345 } });
+
+			expect(consumeTopicSpy).toHaveBeenCalledWith(
+				expect.anything(),
+				expect.objectContaining({ errorRetryDelay: 12345 }),
+			);
+		});
+	});
+
+	describe('fatal consumer errors', () => {
+		/** The logger the node handed the library, which is where fatal errors surface. */
+		function libraryLogger(consumer: FakeConsumer) {
+			const logger = consumer.config.kafkaJS?.logger;
+			if (!logger) throw new Error('the node gave the library no logger');
+			return logger;
+		}
+
+		it('surfaces a non-recoverable consumer error through emitError, as v1 does', async () => {
+			const { emitError } = await startTrigger('v2-fatal');
+			const consumer = await lastFakeConsumer();
+
+			libraryLogger(consumer).error('Broker: Group authorization failed');
+
+			expect(emitError).toHaveBeenCalledTimes(1);
+			expect(emitError.mock.calls[0][0].message).toMatch(/authorization failed/i);
+		});
+
+		it('stays quiet for an error the library can recover from', async () => {
+			const { emitError } = await startTrigger('v2-recoverable');
+			const consumer = await lastFakeConsumer();
+
+			libraryLogger(consumer).error('Broker transport failure');
+
+			expect(emitError).not.toHaveBeenCalled();
+		});
+
+		it('stays quiet for an error caused by our own teardown', async () => {
+			const { emitError, close } = await startTrigger('v2-fatal-on-close');
+			const consumer = await lastFakeConsumer();
+
+			await close();
+			libraryLogger(consumer).error('Broker: Group authorization failed');
+
+			expect(emitError).not.toHaveBeenCalled();
+		});
+	});
+
+	describe('message shape options', () => {
+		it('parses JSON and returns only the message when both are set', async () => {
+			const jsonData = { foo: 'bar' };
+			const { emit } = await startTrigger('v2-json', {
+				options: { jsonParseMessage: true, onlyMessage: true },
+			});
+			const consumer = await lastFakeConsumer();
+
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [{ value: Buffer.from(JSON.stringify(jsonData)) }],
+			});
+
+			expect(emit).toHaveBeenCalledWith([[{ json: jsonData }]]);
+		});
+
+		it('includes headers when returnHeaders is true', async () => {
+			const { emit } = await startTrigger('v2-headers', { options: { returnHeaders: true } });
+			const consumer = await lastFakeConsumer();
+
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [
+					{
+						value: Buffer.from('test-message'),
+						headers: { 'content-type': Buffer.from('application/json') },
+					},
+				],
+			});
+
+			expect(emit).toHaveBeenCalledWith([
+				[
+					{
+						json: {
+							message: 'test-message',
+							topic: 'test-topic',
+							headers: { 'content-type': 'application/json' },
+						},
+					},
+				],
+			]);
+		});
+
+		it('keeps binary data when keepBinaryData is enabled', async () => {
+			const { emit } = await startTrigger('v2-binary', { options: { keepBinaryData: true } });
+			const consumer = await lastFakeConsumer();
+
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [{ value: Buffer.from('binary-data') }],
+			});
+
+			const emittedItem = emit.mock.calls[0][0][0][0];
+			expect(emittedItem).toHaveProperty('binary');
+			expect(emittedItem.json.message).toBe('binary-data');
+		});
+	});
+
+	describe('Schema Registry', () => {
+		it('decodes through the registry when enabled', async () => {
+			const mockDecode = vi.fn().mockResolvedValue({ data: 'decoded-data' });
+			(SchemaRegistry as unknown as Mock).mockImplementation(function () {
+				return { decode: mockDecode } as unknown as Mocked<SchemaRegistry>;
+			});
+
+			const { emit } = await startTrigger('v2-registry', {
+				useSchemaRegistry: true,
+				schemaRegistryUrl: 'http://localhost:8081',
+			});
+			const consumer = await lastFakeConsumer();
+
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [{ value: Buffer.from('avro-encoded') }],
+			});
+
+			expect(SchemaRegistry).toHaveBeenCalledWith({ host: 'http://localhost:8081' });
+			expect(mockDecode).toHaveBeenCalledWith(Buffer.from('avro-encoded'));
+			expect(emit).toHaveBeenCalledWith([
+				[{ json: { message: { data: 'decoded-data' }, topic: 'test-topic' } }],
+			]);
+		});
+
+		it('activates anyway and warns when the registry is unreachable, same as v1', async () => {
+			(SchemaRegistry as unknown as Mock).mockImplementationOnce(function () {
+				throw Object.assign(new Error('connect ECONNREFUSED'), { status: 503 });
+			});
+
+			const { emit, logger } = await startTrigger('v2-registry-down', {
+				useSchemaRegistry: true,
+				schemaRegistryUrl: 'http://localhost:8081',
+			});
+
+			expect(logger.warn).toHaveBeenCalledWith('Could not connect to Schema Registry', {
+				message: 'connect ECONNREFUSED',
+				status: 503,
+			});
+
+			const consumer = await lastFakeConsumer();
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [{ value: Buffer.from('raw-message') }],
+			});
+
+			// No registry to decode with, so the raw message is emitted, as v1 does.
+			expect(emit).toHaveBeenCalledWith([
+				[{ json: { message: 'raw-message', topic: 'test-topic' } }],
+			]);
+		});
+
+		it('fails activation when the registry credential is misconfigured, same as v1', async () => {
+			await expect(
+				testTriggerNode(new KafkaTriggerV2(baseDescription), {
+					mode: 'trigger',
+					node: {
+						credentials: {
+							kafka: { id: '1', name: 'Kafka account' },
+							schemaRegistryApi: { id: '2', name: 'Schema Registry account' },
+						},
+						parameters: {
+							topic: 'test-topic',
+							groupId: 'v2-registry-misconfigured',
+							useSchemaRegistry: true,
+							schemaRegistryUrl: '',
+							resolveOffset: 'immediately',
+						},
+					},
+					credentials: {
+						kafka: credential,
+						schemaRegistryApi: {
+							url: 'https://schema-registry.local:8081',
+							authentication: 'basicAuth',
+							username: 'registry-user',
+							password: '',
+						},
+					},
+				}),
+			).rejects.toThrow('Username and password are required for Schema Registry Basic Auth');
+		});
+	});
+
+	describe('offset resolution', () => {
+		it('waits for the execution before resolving the offset on onCompletion', async () => {
+			const { emit } = await startTrigger('v2-on-completion', { resolveOffset: 'onCompletion' });
+			const consumer = await lastFakeConsumer();
+
+			const delivered = consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [{ value: Buffer.from('message') }],
+			});
+			await new Promise((resolve) => setImmediate(resolve));
+
+			// The offset must not advance while the execution is still running.
+			expect(emit).toHaveBeenCalled();
+			expect(consumer.payloadSpies.resolveOffset).not.toHaveBeenCalled();
+
+			const deferred = emit.mock.calls[0][2];
+			expect(deferred).toBeDefined();
+			deferred?.resolve(mock<IRun>({ status: 'success' }));
+			await delivered;
+
+			expect(consumer.payloadSpies.resolveOffset).toHaveBeenCalledTimes(1);
+		});
+
+		it('does not wait for the execution on immediately', async () => {
+			const { emit } = await startTrigger('v2-immediately');
+			const consumer = await lastFakeConsumer();
+
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [{ value: Buffer.from('message') }],
+			});
+
+			// No deferred promise means nothing to wait on.
+			expect(emit.mock.calls[0][2]).toBeUndefined();
+			expect(consumer.payloadSpies.resolveOffset).toHaveBeenCalledTimes(1);
+		});
+
+		it('leaves the offset unresolved when the execution status is not allowed', async () => {
+			const { emit } = await startTrigger('v2-on-status', {
+				resolveOffset: 'onStatus',
+				allowedStatuses: ['success'],
+				// A rejected status paces the re-delivery before reporting failure;
+				// the default 5s would outlast the test.
+				options: { errorRetryDelay: 1 },
+			});
+			const consumer = await lastFakeConsumer();
+
+			const delivered = consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [{ value: Buffer.from('message') }],
+			});
+			await new Promise((resolve) => setImmediate(resolve));
+
+			emit.mock.calls[0][2]?.resolve(mock<IRun>({ status: 'error' }));
+			await delivered;
+
+			expect(consumer.payloadSpies.resolveOffset).not.toHaveBeenCalled();
+		});
+	});
+
+	describe('manual test run isolation from production', () => {
+		async function startManualRun(parameters: Record<string, unknown> = {}) {
+			const started = await testTriggerNode(new KafkaTriggerV2(baseDescription), {
+				mode: 'manual',
+				node: {
+					parameters: {
+						topic: 'test-topic',
+						groupId: 'orders-consumer',
+						useSchemaRegistry: false,
+						...parameters,
+					},
+				},
+				credential,
+			});
+			await started.manualTriggerFunction?.();
+			return started;
+		}
+
+		it('tells the user which group ACL prefix to grant when the broker refuses it', async () => {
+			// On a cluster with an authorizer, group ACLs are usually granted LITERAL on
+			// the exact Group ID, so the throwaway group is a resource nobody authorized
+			// and only test runs break. The broker names neither the group nor the fix.
+			const { emitError } = await startManualRun();
+			const consumer = await lastFakeConsumer();
+			const logger = consumer.config.kafkaJS?.logger;
+			if (!logger) throw new Error('the node gave the library no logger');
+
+			logger.error('Broker: Group authorization failed');
+
+			expect(emitError).toHaveBeenCalledTimes(1);
+			expect((emitError.mock.calls[0][0] as UserError).description).toContain(
+				'orders-consumer-n8n-manual-',
+			);
+		});
+
+		it('closes a consumer that finished starting while the run was being cancelled', async () => {
+			// Only manual runs can reach this: an activated workflow awaits the start
+			// before n8n has a close function to call. Here the start is handed over as
+			// manualTriggerFunction, so cancelling mid-start used to find no handle yet
+			// and leave a connected consumer behind with nothing holding it.
+			const started = await testTriggerNode(new KafkaTriggerV2(baseDescription), {
+				mode: 'manual',
+				node: {
+					parameters: { topic: 'test-topic', groupId: 'orders-consumer', useSchemaRegistry: false },
+				},
+				credential,
+			});
+
+			// Deliberately not awaited, so close lands while connect/subscribe/run are
+			// still in flight.
+			const starting = started.manualTriggerFunction?.();
+			await started.close?.();
+			await starting;
+
+			const consumer = await lastFakeConsumer();
+			expect(consumer.disconnect).toHaveBeenCalled();
+		});
+
+		it('joins a throwaway group, never the one the activated workflow uses', async () => {
+			await startManualRun();
+
+			const consumer = await lastFakeConsumer();
+			const { groupId } = consumer.config.kafkaJS ?? {};
+			// Sharing the group would let this run commit offsets for messages the
+			// activated workflow never received.
+			expect(groupId).not.toBe('orders-consumer');
+			expect(groupId).toMatch(/^orders-consumer-n8n-manual-.+/);
+		});
+
+		it('waits for the next message rather than replaying the topic', async () => {
+			// Read Messages From Beginning defaults to on, and the throwaway group has
+			// no committed offset, so honouring it would replay everything.
+			await startManualRun({ options: { fromBeginning: true } });
+
+			const consumer = await lastFakeConsumer();
+			expect(consumer.config.kafkaJS?.fromBeginning).toBe(false);
+		});
+
+		it('still honours Read Messages From Beginning for an activated workflow', async () => {
+			await startTrigger('v2-from-beginning', { options: { fromBeginning: true } });
+
+			const consumer = await lastFakeConsumer();
+			expect(consumer.config.kafkaJS?.fromBeginning).toBe(true);
+		});
+	});
+
+	describe('manual test run', () => {
+		it('starts the loop only when the test event is requested, and never waits', async () => {
+			// v1 forces `immediately` in manual mode. The editor discards the run once
+			// it has its sample, so a wait would hold the batch open until close.
+			const { emit, manualTriggerFunction } = await testTriggerNode(
+				new KafkaTriggerV2(baseDescription),
+				{
+					mode: 'manual',
+					node: {
+						parameters: {
+							topic: 'test-topic',
+							groupId: 'v2-manual',
+							useSchemaRegistry: false,
+							resolveOffset: 'onCompletion',
+						},
+					},
+					credential,
+				},
+			);
+
+			expect(getFakeConsumers()).toHaveLength(0);
+
+			await manualTriggerFunction?.();
+
+			const consumer = await lastFakeConsumer();
+			expect(consumer.connect).toHaveBeenCalledTimes(1);
+			expect(emit).not.toHaveBeenCalled();
+
+			await consumer.deliverBatch({
+				topic: 'test-topic',
+				messages: [{ value: Buffer.from('test') }],
+			});
+
+			expect(emit).toHaveBeenCalledWith([[{ json: { message: 'test', topic: 'test-topic' } }]]);
+			// Forced to immediately despite the node asking for onCompletion.
+			expect(emit.mock.calls[0][2]).toBeUndefined();
+		});
+	});
+
+	describe('close', () => {
+		it('disconnects the consumer', async () => {
+			const { close } = await startTrigger('v2-close');
+			const consumer = await lastFakeConsumer();
+
+			await close();
+
+			expect(consumer.disconnect).toHaveBeenCalled();
+		});
+
+		it('reports a failed teardown as a TriggerCloseError, as v1 does', async () => {
+			const { close } = await startTrigger('v2-close-fails');
+			const consumer = await lastFakeConsumer();
+			const teardownError = new Error('The coordinator is not aware of this member');
+			consumer.disconnect.mockRejectedValueOnce(teardownError);
+
+			const error = await close().then(
+				() => null,
+				(e: unknown) => e,
+			);
+
+			expect(error).toBeInstanceOf(TriggerCloseError);
+			expect((error as TriggerCloseError).cause).toBe(teardownError);
+			expect((error as TriggerCloseError).level).toBe('warning');
+		});
+	});
+});

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

@@ -159,23 +159,44 @@ describe('createMessageParser', () => {
 			expect(registry.decode).toHaveBeenCalledWith(Buffer.from('{"a":1}'));
 		});
 
-		it('falls back to the original value and warns when decoding fails', async () => {
+		it('fails the message rather than handing the workflow undecoded bytes', async () => {
+			// v1 warns and passes the raw value through, then commits the offset, so an
+			// unreadable Avro message is lost for good. Throwing leaves the chunk
+			// unresolved instead, and Kafka redelivers it once the registry is healthy.
 			const registry = mock<SchemaRegistry>({
 				decode: vi.fn(async () => {
 					throw new Error('Request failed with status code 404 for https://user:pw@registry');
 				}),
 			});
 
-			expect(await parse({}, message(), registry)).toStrictEqual({
-				json: { message: '{"a":1}', topic: TOPIC },
-			});
-			expect(logger.warn).toHaveBeenCalledWith(
-				'Could not decode message with Schema Registry, returning original message',
+			await expect(parse({}, message(), registry)).rejects.toThrow(
+				'Request failed with status code 404',
+			);
+			expect(logger.error).toHaveBeenCalledWith(
+				'Could not decode message with Schema Registry, leaving it unread',
 				// The sanitizer redacts the URL userinfo the registry client embeds.
 				{ message: 'Request failed with status code 404 for https://***@registry' },
 			);
 		});
 
+		it('throws a sanitized error, since the consume loop logs whatever it is handed', async () => {
+			// The line above scrubs what this file logs, but the throw travels on: the
+			// consume loop logs the error when it leaves the chunk unresolved, so an
+			// unsanitized rethrow would put the registry password straight back in the
+			// log. The original is not attached as `cause` for the same reason.
+			const registry = mock<SchemaRegistry>({
+				decode: vi.fn(async () => {
+					throw new Error('Request failed for https://user:sup3r-secret@registry');
+				}),
+			});
+
+			const thrown = await parse({}, message(), registry).catch((error: Error) => error);
+
+			expect(thrown.message).not.toContain('sup3r-secret');
+			expect(thrown.message).toContain('https://***@registry');
+			expect(thrown.cause).toBeUndefined();
+		});
+
 		it('does not call the registry for an empty value', async () => {
 			const registry = mock<SchemaRegistry>({ decode: vi.fn() });
 

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

@@ -0,0 +1,122 @@
+import { ensureError } from '@n8n/utils/errors/ensure-error';
+import type {
+	INodeTypeBaseDescription,
+	INodeTypeDescription,
+	INodeType,
+	ITriggerFunctions,
+	ITriggerResponse,
+} from 'n8n-workflow';
+import { NodeOperationError, TriggerCloseError } from 'n8n-workflow';
+
+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';
+
+export class KafkaTriggerV2 implements INodeType {
+	description: INodeTypeDescription;
+
+	constructor(baseDescription: INodeTypeBaseDescription) {
+		this.description = {
+			...baseDescription,
+			...versionDescription,
+		};
+	}
+
+	async trigger(this: ITriggerFunctions): Promise<ITriggerResponse> {
+		const settings = getSettings.call(this);
+		const credentials = await this.getCredentials<KafkaCredentials>('kafka');
+
+		// Resolved before the consumer connects: a bad Schema Registry credential
+		// must fail activation rather than leave a connected consumer behind. Shared
+		// with v1: config/credential problems (NodeOperationError) fail activation
+		// loudly, but a registry that is merely unreachable only logs a warning, so
+		// a transient outage does not block the trigger from starting.
+		const registry = await setSchemaRegistry(this);
+
+		const parseMessage = createMessageParser(
+			settings.parser,
+			this.logger,
+			registry,
+			this.helpers.prepareBinaryData,
+		);
+
+		// Aborted before the consumer disconnects, so an execution the emitter is
+		// still waiting on cannot hold teardown open.
+		const closeController = new AbortController();
+		const emit = createDataEmitter(this, settings.emitter, closeController.signal);
+
+		let handle: KafkaConsumerHandle | undefined;
+		// A manual run starts the consumer from `manualTriggerFunction`, so unlike an
+		// activated workflow the start is not awaited before n8n can call close.
+		// Cancelling the test run mid-start would otherwise find `handle` still unset
+		// and leave a consumer connected with nothing left holding it.
+		let startup: Promise<void> | undefined;
+
+		const startConsumer = async () => {
+			startup = startConsumerOnce();
+			await startup;
+		};
+
+		const startConsumerOnce = async () => {
+			try {
+				const consumer = await createKafkaConsumer(credentials, settings.consumer, {
+					logger: this.logger,
+					// v1 routes non-restartable consumer crashes to emitError so n8n
+					// re-activates the trigger. Errors caused by our own teardown are
+					// not failures, so they stay quiet.
+					onFatalError: (error) => {
+						if (closeController.signal.aborted) return;
+						this.emitError(
+							explainManualRunGroupDenial(error, settings.configuredGroupId, settings.isManualRun),
+						);
+					},
+				});
+
+				handle = await consumeTopic(consumer, {
+					topic: settings.topic,
+					parseMessage,
+					emit,
+					logger: this.logger,
+					batchSize: settings.batchSize,
+					partitionsConsumedConcurrently: settings.partitionsConsumedConcurrently,
+					errorRetryDelay: settings.errorRetryDelay,
+				});
+			} catch (error) {
+				throw new NodeOperationError(this.getNode(), error);
+			}
+		};
+
+		const closeFunction = async () => {
+			closeController.abort();
+			try {
+				// Let an in-flight start finish so its consumer is closed rather than
+				// leaked. Its own failure is not a teardown failure, and it has already
+				// been reported to whoever awaited the start.
+				await startup?.catch(() => {});
+				await handle?.close();
+			} catch (error) {
+				// A disconnect that overruns its bound is reported the way v1 reports
+				// teardown failures, rather than as an unattributed rejection. It happens
+				// in practice: a consumer fenced by the processing deadline does not
+				// disconnect within the bound.
+				throw new TriggerCloseError(this.getNode(), {
+					cause: ensureError(error),
+					level: 'warning',
+				});
+			}
+		};
+
+		if (this.getMode() !== 'manual') {
+			await startConsumer();
+			return { closeFunction };
+		}
+
+		return {
+			closeFunction,
+			manualTriggerFunction: startConsumer,
+		};
+	}
+}

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

@@ -0,0 +1,307 @@
+/* eslint-disable n8n-nodes-base/node-filename-against-convention */
+// The node class lives in KafkaTriggerV2.node.ts; this file only holds its UI
+// description, so the filename cannot match `description.name` as the rule
+// expects. Same exemption Notion, NocoDB and Webflow take for the same split.
+import type { INodeTypeDescription } from 'n8n-workflow';
+import { NodeConnectionTypes } from 'n8n-workflow';
+
+export const versionDescription: INodeTypeDescription = {
+	displayName: 'Kafka Trigger',
+	name: 'kafkaTrigger',
+	icon: { light: 'file:kafka.svg', dark: 'file:kafka.dark.svg' },
+	group: ['trigger'],
+	version: 2,
+	description: 'Consume messages from a Kafka topic',
+	defaults: {
+		name: 'Kafka Trigger',
+	},
+	inputs: [],
+	outputs: [NodeConnectionTypes.Main],
+	credentials: [
+		{
+			name: 'kafka',
+			required: true,
+		},
+		{
+			name: 'schemaRegistryApi',
+			required: false,
+			displayName: 'Schema Registry',
+			displayOptions: {
+				show: {
+					useSchemaRegistry: [true],
+				},
+			},
+		},
+	],
+	properties: [
+		{
+			displayName: 'Topic',
+			name: 'topic',
+			type: 'string',
+			default: '',
+			required: true,
+			placeholder: 'topic-name',
+			description: 'Name of the queue of topic to consume from',
+		},
+		{
+			displayName: 'Group ID',
+			name: 'groupId',
+			type: 'string',
+			default: '',
+			required: true,
+			placeholder: 'n8n-kafka',
+			description: 'ID of the consumer group',
+		},
+		{
+			displayName: 'Resolve Offset',
+			name: 'resolveOffset',
+			type: 'options',
+			default: 'onCompletion',
+			description:
+				'Select on which condition the offsets should be resolved. In the manual mode, when execution started by clicking on Execute Workflow or Execute Step button, offsets are always resolved immediately after message received.',
+			options: [
+				{
+					name: 'On Execution Completion',
+					value: 'onCompletion',
+					description: 'Resolve offset after execution completion regardless of the status',
+				},
+				{
+					name: 'On Execution Success',
+					value: 'onSuccess',
+					description: 'Resolve offset only if execution status equals success',
+				},
+				{
+					name: 'On Allowed Execution Statuses',
+					value: 'onStatus',
+					description: 'Resolve offset only if execution status in the list of selected statuses',
+				},
+				{
+					name: 'Immediately',
+					value: 'immediately',
+					description:
+						'Resolve offset immediately after message received. This option is not recommended as it can cause messages loss.',
+				},
+			],
+		},
+		{
+			displayName: 'Allowed Statuses',
+			name: 'allowedStatuses',
+			type: 'multiOptions',
+			default: ['success'],
+			options: [
+				{
+					name: 'Canceled',
+					value: 'canceled',
+				},
+				{
+					name: 'Crashed',
+					value: 'crashed',
+				},
+				{
+					name: 'Error',
+					value: 'error',
+				},
+				{
+					name: 'New',
+					value: 'new',
+				},
+				{
+					name: 'Running',
+					value: 'running',
+				},
+				{
+					name: 'Success',
+					value: 'success',
+				},
+				{
+					name: 'Unknown',
+					value: 'unknown',
+				},
+				{
+					name: 'Waiting',
+					value: 'waiting',
+				},
+			],
+			displayOptions: {
+				show: {
+					resolveOffset: ['onStatus'],
+				},
+			},
+		},
+		{
+			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: 'Options',
+			name: 'options',
+			type: 'collection',
+			default: {},
+			placeholder: 'Add option',
+			options: [
+				// "Allow Topic Creation" is deliberately absent. v1 declares it and never
+				// reads it, and wiring it to the consumer's `allow.auto.create.topics`
+				// changed nothing: measured against a real broker with auto-creation
+				// enabled, a consumer subscribed to a missing topic did not create it
+				// with the flag on or off. Not shipped as a third control that does
+				// nothing.
+				{
+					displayName: 'Auto Commit Interval',
+					name: 'autoCommitInterval',
+					type: 'number',
+					default: 0,
+					description:
+						'The consumer will commit offsets after a given period, for example, five seconds',
+					hint: 'Value in milliseconds',
+				},
+				{
+					displayName: 'Batch Size',
+					name: 'batchSize',
+					type: 'number',
+					default: 1,
+					description:
+						'Number of messages to process in each batch, when set to 1, message-by-message processing is enabled',
+				},
+				// "Each Batch Auto Resolve" is deliberately absent. The consume loop
+				// resolves offsets chunk by chunk and turns the library's automatic
+				// resolution off, so honouring it would mark messages read that no
+				// execution ever saw. Dropped rather than shipped as a dead control.
+				{
+					displayName: 'Fetch Max Bytes',
+					name: 'fetchMaxBytes',
+					type: 'number',
+					default: 1048576,
+					description:
+						'Maximum amount of data the server should return for a fetch request. In bytes. Default is 1MB. Higher values allow fetching more messages at once.',
+				},
+				{
+					displayName: 'Fetch Min Bytes',
+					name: 'fetchMinBytes',
+					type: 'number',
+					default: 1,
+					description:
+						'Minimum amount of data the server should return for a fetch request. In bytes. Server will wait up to fetchMaxWaitTime for this amount to accumulate.',
+				},
+				{
+					displayName: 'Heartbeat Interval',
+					name: 'heartbeatInterval',
+					type: 'number',
+					default: 10000,
+					description:
+						'Controls how often the consumer sends heartbeats to the broker to indicate it is still alive. Must be lower than Session Timeout. Recommended value is approximately one third of the Session Timeout (for example: 10s heartbeat with 30s session timeout).',
+					hint: 'Value in milliseconds. Lowered automatically if it is more than a third of the Session Timeout.',
+				},
+				{
+					displayName: 'Max Number of Requests',
+					name: 'maxInFlightRequests',
+					type: 'number',
+					default: 1,
+					description:
+						'The maximum number of unacknowledged requests the client will send on a single connection',
+				},
+				{
+					displayName: 'Read Messages From Beginning',
+					name: 'fromBeginning',
+					type: 'boolean',
+					default: true,
+					description: 'Whether to read message from beginning',
+				},
+				{
+					displayName: 'JSON Parse Message',
+					name: 'jsonParseMessage',
+					type: 'boolean',
+					default: false,
+					description: 'Whether to try to parse the message to an object',
+				},
+				{
+					displayName: 'Keep Message as Binary Data',
+					name: 'keepBinaryData',
+					type: 'boolean',
+					default: false,
+					description:
+						'Whether to keep message value as binary data for downstream processing (e.g., Avro deserialization)',
+				},
+				{
+					displayName: 'Partitions Consumed Concurrently',
+					name: 'partitionsConsumedConcurrently',
+					type: 'number',
+					default: 0,
+					description:
+						'Number of Kafka partitions to process in parallel. Controls how many partitions are processed concurrently by the consumer.',
+					hint: 'Set to 0 to process all partitions sequentially',
+				},
+				{
+					displayName: 'Only Message',
+					name: 'onlyMessage',
+					type: 'boolean',
+					displayOptions: {
+						show: {
+							jsonParseMessage: [true],
+						},
+					},
+					default: false,
+					description: 'Whether to return only the message property',
+				},
+				{
+					displayName: 'Return Headers',
+					name: 'returnHeaders',
+					type: 'boolean',
+					default: false,
+					description: 'Whether to return the headers received from Kafka',
+				},
+				{
+					displayName: 'Rebalance Timeout',
+					name: 'rebalanceTimeout',
+					type: 'number',
+					default: 600000,
+					description:
+						'How long one batch may take to process before the consumer is dropped from its group. Only used when the workflow has no execution timeout of its own, since that timeout is the real deadline and takes precedence.',
+					hint: 'Value in milliseconds',
+				},
+				{
+					displayName: 'Retry Delay on Error',
+					name: 'errorRetryDelay',
+					type: 'number',
+					default: 5000,
+					description:
+						'Delay in milliseconds before retrying after a failed offset resolution. This prevents rapid retry loops that could overwhelm the Kafka broker.',
+					hint: 'Value in milliseconds',
+					typeOptions: {
+						minValue: 1000,
+					},
+					displayOptions: {
+						hide: {
+							'/resolveOffset': ['immediately'],
+						},
+					},
+				},
+				{
+					displayName: 'Session Timeout',
+					name: 'sessionTimeout',
+					type: 'number',
+					default: 30000,
+					description:
+						'Timeout in milliseconds used to detect failures. Has to be higher than Heartbeat Interval. During the workflow execution heartbeat will be sent periodically to keep the session alive with configured Heartbeat Interval.',
+					hint: 'Value in milliseconds. Lowering this below three times the Heartbeat Interval will lower the heartbeat to match.',
+				},
+			],
+		},
+	],
+};

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

@@ -0,0 +1,378 @@
+import { randomUUID } from 'node:crypto';
+import type { ITriggerFunctions, Logger } from 'n8n-workflow';
+import { UserError } from 'n8n-workflow';
+
+import { DEFAULT_EXECUTION_TIMEOUT_SECONDS } from './consumer';
+import type { DataEmitterOptions, KafkaMessageParserOptions, ResolveOffsetMode } from './consumer';
+import type { KafkaConsumerOptions } from './transport';
+
+/** v1.3's `options` collection. Every field is optional: a `collection` node
+ * parameter only carries the keys the user actually set, never its declared
+ * UI defaults, so the converters below resolve each one explicitly. */
+export interface KafkaTriggerV2Options extends KafkaMessageParserOptions {
+	batchSize?: number;
+	partitionsConsumedConcurrently?: number;
+	errorRetryDelay?: number;
+	sessionTimeout?: number;
+	heartbeatInterval?: number;
+	rebalanceTimeout?: number;
+	fetchMaxBytes?: number;
+	fetchMinBytes?: number;
+	maxInFlightRequests?: number;
+	fromBeginning?: boolean;
+	autoCommitInterval?: number;
+}
+
+/**
+ * Everything `trigger()` needs, already read, defaulted and converted. The
+ * node's own code only ever deals with `settings.xxx`, never a raw
+ * `getNodeParameter` call or an `options.foo as bar`.
+ */
+export interface KafkaTriggerSettings {
+	topic: string;
+	/** Whether this is an editor test run, which gets a throwaway consumer group. */
+	isManualRun: boolean;
+	/** The Group ID as typed, before a manual run's suffix. For error messages. */
+	configuredGroupId: string;
+	/** Ready for the consumer factory, including the manual-run group. */
+	consumer: KafkaConsumerOptions;
+	/** Ready for the data emitter, including the manual-run offset mode. */
+	emitter: DataEmitterOptions;
+	/** The subset of options that changes the parsed item shape. */
+	parser: KafkaMessageParserOptions;
+	batchSize?: number;
+	partitionsConsumedConcurrently?: number;
+	errorRetryDelay?: number;
+}
+
+/** v1's defaults for the consumer settings, from `createConsumerConfig`. v2 is
+ * always >= 1.3, so the heartbeat default is 1.3's 10s rather than 3s. */
+const DEFAULT_SESSION_TIMEOUT_MS = 30_000;
+const DEFAULT_HEARTBEAT_INTERVAL_MS = 10_000;
+
+/**
+ * The most we may ask for. librdkafka accepts 1..86400000 for
+ * `max.poll.interval.ms`, and the library doubles whatever it is given, so half
+ * of that ceiling is the limit here.
+ *
+ * Going over does not fail cleanly: the doubled value overflows the 32-bit int
+ * librdkafka stores it in, and the consumer refuses to connect complaining
+ * about a negative number nobody chose. Measured against a real broker with a
+ * 30 day workflow timeout: `value -1702967296 is outside allowed range
+ * 1..86400000`.
+ */
+const MAX_REBALANCE_TIMEOUT_MS = 43_200_000;
+
+/** librdkafka's range for `auto.commit.interval.ms`. 0 turns interval commits off. */
+const MAX_AUTO_COMMIT_INTERVAL_MS = 86_400_000;
+
+/**
+ * Kafka's own guidance, and the ratio both defaults already sit at: a heartbeat
+ * every 10s against a 30s session. Three beats fit inside one session, so two
+ * can be lost before the broker gives up on the consumer.
+ */
+const HEARTBEATS_PER_SESSION = 3;
+
+/**
+ * Keeps the heartbeat frequent enough for the session timeout it is paired with.
+ *
+ * The two options are independent in the UI but not in Kafka, and getting them
+ * wrong fails silently rather than loudly. Lower Session Timeout to 10s and
+ * leave the 10s heartbeat default alone, and the first beat lands exactly on the
+ * deadline: the broker fences the consumer, the uncommitted offset is lost, and
+ * the same message is redelivered forever with no error anywhere. Measured
+ * against a real broker: a workflow that takes 5s re-ran the same message every
+ * ~10s indefinitely.
+ *
+ * v1 has the same trap. Clamping is a deliberate improvement over it, and the
+ * safe direction is unambiguous, so it is applied rather than only warned about.
+ * @param heartbeatInterval - Resolved Heartbeat Interval, in milliseconds
+ * @param sessionTimeout - Resolved Session Timeout, in milliseconds
+ * @param logger - Warns when the supplied heartbeat had to be lowered
+ */
+function heartbeatWithinSession(
+	heartbeatInterval: number,
+	sessionTimeout: number,
+	logger?: Logger,
+): number {
+	// An unusable session timeout is left to librdkafka to reject by name.
+	if (!Number.isFinite(sessionTimeout) || sessionTimeout <= 0) return heartbeatInterval;
+
+	const largest = Math.floor(sessionTimeout / HEARTBEATS_PER_SESSION);
+	if (Number.isFinite(heartbeatInterval) && heartbeatInterval <= largest) {
+		return heartbeatInterval;
+	}
+
+	logger?.warn(
+		'Kafka Heartbeat Interval lowered to stay under a third of the Session Timeout, so the consumer is not dropped from its group',
+		{ supplied: heartbeatInterval, applied: largest, sessionTimeout },
+	);
+	return largest;
+}
+
+/**
+ * A user-supplied millisecond value, or `undefined` to leave the library's own
+ * default in place. Options can come from an expression, so a value that
+ * librdkafka would reject is dropped with a warning rather than passed on: it
+ * fails the whole connection, and the error names a number nobody typed.
+ * @param value - The raw option
+ * @param max - Largest value librdkafka accepts
+ * @param label - Option name, for the warning
+ * @param logger - Warns when a supplied value could not be used
+ */
+function millisecondsInRange(
+	value: number | undefined,
+	max: number,
+	label: string,
+	logger?: Logger,
+): number | undefined {
+	if (value === undefined) return undefined;
+
+	const usable = Number.isFinite(value) && value >= 0 && value <= max;
+	if (!usable) {
+		logger?.warn(`Kafka ${label} ignored, outside the range the library accepts`, {
+			supplied: value,
+			allowed: `0..${max}`,
+		});
+		return undefined;
+	}
+
+	return value;
+}
+
+/**
+ * Reads and resolves every node parameter up front, so the trigger below never
+ * touches one directly. The single cast is contained here: a `collection`
+ * parameter is untyped at the boundary, and this is the one place that knows
+ * its shape.
+ */
+export function getSettings(this: ITriggerFunctions): KafkaTriggerSettings {
+	const options = this.getNodeParameter('options', {}) as KafkaTriggerV2Options;
+	const { executionTimeout } = this.getWorkflowSettings();
+
+	// A manual test run always resolves immediately, as in v1: the editor discards
+	// the run once it has its sample, so waiting on an execution that nothing will
+	// finish would hold the batch open until close.
+	const isManualRun = this.getMode() === 'manual';
+	const resolveOffsetMode = isManualRun
+		? 'immediately'
+		: // Falls back to the field's own default rather than v1's 'immediately', so
+			// an absent value keeps the at-least-once behaviour.
+			(this.getNodeParameter('resolveOffset', 'onCompletion') as ResolveOffsetMode);
+
+	const configuredGroupId = this.getNodeParameter('groupId') as string;
+
+	return {
+		topic: this.getNodeParameter('topic') as string,
+		isManualRun,
+		configuredGroupId,
+		consumer: toConsumerOptions(
+			// Read Messages From Beginning defaults to on, and a manual run's group is
+			// brand new, so honouring it would replay the whole topic into the editor.
+			// On an activated workflow the setting is moot anyway, since the group
+			// already has a committed offset to resume from.
+			isManualRun ? { ...options, fromBeginning: false } : options,
+			manualRunGroupId(configuredGroupId, isManualRun),
+			executionTimeout,
+			this.logger,
+		),
+		emitter: toEmitterOptions(
+			options,
+			resolveOffsetMode,
+			this.getNodeParameter('allowedStatuses', []) as string[],
+			executionTimeout,
+		),
+		parser: options,
+		batchSize: options.batchSize,
+		// 0 means "all partitions sequentially", which the loop expresses by
+		// leaving the key off and taking its own default of 1.
+		partitionsConsumedConcurrently: options.partitionsConsumedConcurrently || undefined,
+		errorRetryDelay: options.errorRetryDelay,
+	};
+}
+
+/**
+ * The consumer group a run joins. A manual run gets a throwaway one.
+ *
+ * Sharing the configured group with an activated workflow means Kafka splits the
+ * partitions between the two, and a test run resolves offsets immediately, so
+ * pressing "Listen for test event" would mark messages read that the activated
+ * workflow never saw. A group of its own leaves production untouched. v1 has
+ * this defect and needs its own fix.
+ *
+ * The throwaway group is left behind on the broker; Kafka expires an empty group
+ * on its own once its retention window passes.
+ * @param configured - The node's Group ID parameter
+ * @param isManualRun - Whether this is an editor test run
+ */
+export function manualRunGroupId(configured: string, isManualRun: boolean): string {
+	return isManualRun ? `${MANUAL_RUN_PREFIX(configured)}${randomUUID()}` : configured;
+}
+
+/** The part of a manual run's group id that is stable, and so grantable in an ACL. */
+const MANUAL_RUN_PREFIX = (configured: string) => `${configured}-n8n-manual-`;
+
+/** Broker rejections that mean the group itself was refused, not the credential. */
+const GROUP_AUTHORIZATION_FAILED = /group authorization failed/i;
+
+/**
+ * Explains a group authorization failure on a manual run, where the group the
+ * broker refused is one the user never chose and cannot see.
+ *
+ * On a cluster with an authorizer, group ACLs are usually granted `LITERAL` on
+ * the exact Group ID. The throwaway group is a different resource name, so the
+ * join is denied while the activated workflow keeps working: "Listen for test
+ * event" fails on its own, and the raw broker message names neither the group
+ * nor the fix. A `PREFIXED` ACL on the stable part covers every future test run.
+ *
+ * Only manual runs are rewritten. On an activated workflow the group is exactly
+ * what the user typed, so the broker's own message is already actionable.
+ * @param error - The error raised from the library's log stream
+ * @param configuredGroupId - The node's Group ID parameter, before the suffix
+ * @param isManualRun - Whether this is an editor test run
+ */
+export function explainManualRunGroupDenial(
+	error: Error,
+	configuredGroupId: string,
+	isManualRun: boolean,
+): Error {
+	if (!isManualRun || !GROUP_AUTHORIZATION_FAILED.test(error.message)) return error;
+
+	return new UserError('Kafka refused the consumer group used for a test run', {
+		description: `A test run uses a throwaway consumer group so it cannot mark messages read for the activated workflow, and this cluster has not authorized it. Grant a prefixed group ACL for "${MANUAL_RUN_PREFIX(configuredGroupId)}", or run the workflow activated instead of testing it.`,
+		cause: error,
+	});
+}
+
+/**
+ * Maps the node's options onto the consumer factory.
+ *
+ * `rebalanceTimeout` is the one value not passed straight through. In this
+ * library it also becomes `max.poll.interval.ms`, the deadline to finish one
+ * batch before the consumer is dropped from its group, and the library doubles
+ * whatever it is given (`_consumer.js:711`). So the workflow's own execution
+ * timeout is the deadline we actually want, and it is halved here to survive
+ * that doubling, then capped at {@link MAX_REBALANCE_TIMEOUT_MS}. An unbounded
+ * workflow timeout (<= 0), or one that is not a usable number, has no deadline
+ * to derive from, so the Rebalance Timeout option stands in.
+ * @param options - The node's `options` collection
+ * @param groupId - The node's consumer group id
+ * @param executionTimeoutSeconds - `getWorkflowSettings().executionTimeout`, raw
+ * @param logger - Warns when a supplied value could not be used as given
+ */
+export function toConsumerOptions(
+	options: KafkaTriggerV2Options,
+	groupId: string,
+	executionTimeoutSeconds: number | undefined,
+	logger?: Logger,
+): KafkaConsumerOptions {
+	// A workflow timeout arrives as a number, but Rebalance Timeout can come from
+	// an expression, so neither is trusted to be a usable one.
+	const fromWorkflow =
+		typeof executionTimeoutSeconds === 'number' &&
+		Number.isFinite(executionTimeoutSeconds) &&
+		executionTimeoutSeconds > 0;
+	const configured = options.rebalanceTimeout;
+	const usableOption =
+		typeof configured === 'number' && Number.isFinite(configured) && configured > 0;
+
+	// Three sources, most specific first.
+	//
+	// The last one used to be the Rebalance Timeout default, which handed the broker
+	// a 10 minute deadline while the emitter was still prepared to wait an hour for
+	// the same execution. Anything in between was fenced and redelivered while n8n
+	// believed the run still owned it: two of our own defaults disagreeing about one
+	// deadline, so they are now the same value by construction.
+	let deadlineMs: number;
+	if (fromWorkflow) {
+		deadlineMs = executionTimeoutSeconds * 1000;
+	} else if (usableOption) {
+		// No usable workflow timeout, but the user set this deliberately.
+		deadlineMs = configured;
+	} else {
+		// Nothing set anywhere, so match how long the emitter will actually wait.
+		deadlineMs = DEFAULT_EXECUTION_TIMEOUT_SECONDS * 1000;
+	}
+
+	const sessionTimeout = options.sessionTimeout ?? DEFAULT_SESSION_TIMEOUT_MS;
+
+	// librdkafka refuses a `max.poll.interval.ms` below `session.timeout.ms`
+	// (`rdkafka_conf.c:4257`, on the classic group protocol we use), and the
+	// deadline becomes exactly that. A range check cannot catch it because both
+	// values are individually legal. Left alone, a workflow timeout under the 30s
+	// session default made the consumer refuse to connect, complaining about two
+	// numbers where the user only chose one.
+	// An unusable session timeout is left for librdkafka to reject by name, as the
+	// heartbeat clamp does. Flooring against it would turn the deadline into NaN
+	// and hide the value actually at fault behind a second complaint.
+	const sessionUsable = Number.isFinite(sessionTimeout) && sessionTimeout > 0;
+	const floored = sessionUsable ? Math.max(deadlineMs, sessionTimeout) : deadlineMs;
+	if (floored !== deadlineMs) {
+		logger?.warn(
+			'Kafka processing deadline raised to the Session Timeout, the shortest the library allows',
+			{ requestedMs: deadlineMs, appliedMs: floored, sessionTimeout },
+		);
+	}
+
+	const wanted = Math.ceil(floored / 2);
+	// Capped last: the ceiling exists to stop the doubled value overflowing a
+	// 32-bit int, which outranks the floor above. They cannot both bite, since a
+	// session timeout large enough to collide is one librdkafka already rejects.
+	const rebalanceTimeout = Math.min(wanted, MAX_REBALANCE_TIMEOUT_MS);
+	if (rebalanceTimeout !== wanted) {
+		logger?.warn('Kafka processing deadline capped at the largest the library accepts, 24 hours', {
+			requestedMs: deadlineMs,
+			appliedMs: rebalanceTimeout * 2,
+		});
+	}
+
+	return {
+		groupId,
+		sessionTimeout,
+		heartbeatInterval: heartbeatWithinSession(
+			options.heartbeatInterval ?? DEFAULT_HEARTBEAT_INTERVAL_MS,
+			sessionTimeout,
+			logger,
+		),
+		rebalanceTimeout,
+		maxBytesPerPartition: options.fetchMaxBytes,
+		minBytes: options.fetchMinBytes,
+		// v1 turns 0 into `null` to mean "no limit". The library has no such
+		// sentinel, so leave the key off and let its own default stand.
+		maxInFlightRequests: options.maxInFlightRequests || undefined,
+		fromBeginning: options.fromBeginning,
+		// 0 is a real setting here, not "unset", so it must not go through the
+		// falsy-to-undefined treatment maxInFlightRequests gets.
+		autoCommitInterval: millisecondsInRange(
+			options.autoCommitInterval,
+			MAX_AUTO_COMMIT_INTERVAL_MS,
+			'Auto Commit Interval',
+			logger,
+		),
+	};
+}
+
+/**
+ * Maps the node's options and offset settings onto the emitter.
+ *
+ * `executionTimeoutSeconds` is passed through raw: n8n treats <= 0 as
+ * explicitly unbounded, and the emitter handles that. Coercing it to a default
+ * here would reintroduce a deadline the user switched off.
+ * @param options - The node's `options` collection
+ * @param resolveOffsetMode - Already resolved, including the manual-mode override
+ * @param allowedStatuses - Only meaningful when the mode is `onStatus`
+ * @param executionTimeoutSeconds - `getWorkflowSettings().executionTimeout`, raw
+ */
+export function toEmitterOptions(
+	options: KafkaTriggerV2Options,
+	resolveOffsetMode: ResolveOffsetMode,
+	allowedStatuses: string[],
+	executionTimeoutSeconds: number | undefined,
+): DataEmitterOptions {
+	return {
+		resolveOffsetMode,
+		allowedStatuses,
+		executionTimeoutSeconds,
+		errorRetryDelay: options.errorRetryDelay,
+	};
+}

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

@@ -40,7 +40,13 @@ export interface DataEmitterOptions {
 // Constants
 // ---------------------------------------------------------------------------
 
-const DEFAULT_EXECUTION_TIMEOUT_SECONDS = 3600;
+/**
+ * The wait an absent workflow execution timeout falls back to. Exported because
+ * the consumer's processing deadline has to agree with it: if the emitter waits
+ * an hour for an execution the broker fenced ten minutes in, the message is
+ * redelivered while n8n still thinks the run owns it.
+ */
+export const DEFAULT_EXECUTION_TIMEOUT_SECONDS = 3600;
 
 // Every hand-off returns one of these two, so they are shared rather than
 // rebuilt. Frozen because a caller mutating the one it was handed would change

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

@@ -8,7 +8,7 @@ import type {
 	ITriggerFunctions,
 	Logger,
 } from 'n8n-workflow';
-import { jsonParse } from 'n8n-workflow';
+import { jsonParse, OperationalError } from 'n8n-workflow';
 
 import { sanitizeRegistryError } from '../../utils';
 
@@ -78,10 +78,27 @@ export function createMessageParser(
 			try {
 				value = await registry.decode(message.value);
 			} catch (error) {
-				logger.warn(
-					'Could not decode message with Schema Registry, returning original message',
-					sanitizeRegistryError(error),
-				);
+				// Deliberately fatal, unlike v1, which warns and hands the workflow the
+				// raw bytes. An undecoded Avro value is a magic byte, a schema id and
+				// binary rendered through UTF-8: unusable downstream, and v1 commits the
+				// offset anyway, so the message is gone for good. Measured against a real
+				// broker: stop the registry, consume, restart it, rejoin the same group,
+				// and the message never comes back.
+				//
+				// Throwing hands it to the consume loop, which paces a retry and leaves
+				// the chunk unresolved, so Kafka redelivers it and the message survives
+				// until the registry is healthy again. The partition stalls meanwhile,
+				// which is the visible, recoverable failure we want in place of silent
+				// corruption. A JSON parse failure above stays a warning: a message that
+				// is not JSON is a legitimate case and the string is still usable.
+				const sanitized = sanitizeRegistryError(error);
+				logger.error('Could not decode message with Schema Registry, leaving it unread', sanitized);
+				// Sanitized, and deliberately without the original as `cause`. A registry
+				// error message can carry the URL it was built from, userinfo included,
+				// and the consume loop logs whatever this throws when it decides to leave
+				// the chunk unresolved. Rethrowing the raw error, or attaching it, would
+				// put the credential back in the log the line above just scrubbed.
+				throw new OperationalError(sanitized.message);
 			}
 		}
 

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

@@ -4,7 +4,7 @@ export {
 	DEFAULT_PARTITIONS_CONSUMED_CONCURRENTLY,
 } from './ConsumeTopic';
 export type { ConsumeTopicOptions, KafkaConsumerHandle } from './ConsumeTopic';
-export { createDataEmitter } from './DataEmitter';
+export { createDataEmitter, DEFAULT_EXECUTION_TIMEOUT_SECONDS } from './DataEmitter';
 export type {
 	DataEmitter,
 	DataEmitterContext,

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

@@ -38,6 +38,11 @@ export interface KafkaConsumerOptions {
 	maxInFlightRequests?: number;
 	/** Start at the earliest offset. Per-consumer here, unlike kafkajs's per-subscribe. */
 	fromBeginning?: boolean;
+	/**
+	 * How often read progress is saved, in ms. librdkafka takes 0..86400000, where
+	 * 0 turns interval commits off. Overrides {@link CONSUMER_DEFAULTS}.
+	 */
+	autoCommitInterval?: number;
 }
 
 /** Wiring for the library's own log output, kept apart from its config keys. */