alighasami
3d5eaf9445
Security: Sync from Public / sync-from-public (push) Has been cancelled
Test: Benchmark Nightly / build (push) Has been cancelled
Test: Benchmark Nightly / Notify Cats on failure (push) Has been cancelled
CI: Python / Checks (push) Has been cancelled
Test: Evals Python / Workflow Comparison Python (push) Has been cancelled
Util: Check Docs URLs / check-docs-urls (push) Has been cancelled
Test: Visual Storybook / Cloudflare Pages (push) Has been cancelled
Test: E2E Performance / build-and-test-performance (push) Has been cancelled
Test: Workflows Nightly / Run Workflow Tests (push) Has been cancelled
Util: Cleanup CI Docker Images / Delete stale CI images (push) Has been cancelled
Test: Benchmark Destroy Env / build (push) Has been cancelled
Util: Update Node Popularity / update-popularity (push) Has been cancelled
Test: E2E Coverage Weekly / Coverage Tests (push) Has been cancelled
2017 lines
51 KiB
TypeScript
2017 lines
51 KiB
TypeScript
import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
|
|
import { mock } from 'jest-mock-extended';
|
|
import {
|
|
Kafka,
|
|
logLevel,
|
|
type Consumer,
|
|
type ConsumerRunConfig,
|
|
type EachBatchHandler,
|
|
type EachMessageHandler,
|
|
type IHeaders,
|
|
type KafkaMessage,
|
|
type RecordBatchEntry,
|
|
} from 'kafkajs';
|
|
import { NodeOperationError, type IRun } from 'n8n-workflow';
|
|
|
|
import { testTriggerNode } from '@test/nodes/TriggerHelpers';
|
|
|
|
import { KafkaTrigger } from '../KafkaTrigger.node';
|
|
|
|
jest.mock('kafkajs');
|
|
jest.mock('@kafkajs/confluent-schema-registry');
|
|
jest.mock('n8n-workflow', () => {
|
|
const actual = jest.requireActual('n8n-workflow');
|
|
return {
|
|
...actual,
|
|
sleep: jest.fn().mockResolvedValue(undefined),
|
|
};
|
|
});
|
|
|
|
describe('KafkaTrigger Node', () => {
|
|
let mockKafka: jest.Mocked<Kafka>;
|
|
let mockRegistry: jest.Mocked<SchemaRegistry>;
|
|
let mockConsumerConnect: jest.Mock;
|
|
let mockConsumerSubscribe: jest.Mock;
|
|
let mockConsumerRun: jest.Mock;
|
|
let mockConsumerDisconnect: jest.Mock;
|
|
let mockConsumerCreate: jest.Mock;
|
|
let mockRegistryDecode: jest.Mock;
|
|
let publishMessage: (message: Partial<KafkaMessage>) => Promise<void>;
|
|
|
|
beforeEach(() => {
|
|
const mockEachMessageHolder = {
|
|
handler: jest.fn(async () => {}) as jest.Mocked<EachMessageHandler>,
|
|
};
|
|
const mockEachBatchHolder = {
|
|
handler: jest.fn(async () => {}) as jest.Mocked<EachBatchHandler>,
|
|
};
|
|
mockConsumerConnect = jest.fn();
|
|
mockConsumerSubscribe = jest.fn();
|
|
mockConsumerRun = jest.fn(({ eachMessage, eachBatch }: ConsumerRunConfig) => {
|
|
if (eachMessage) {
|
|
mockEachMessageHolder.handler = eachMessage;
|
|
}
|
|
if (eachBatch) {
|
|
mockEachBatchHolder.handler = eachBatch;
|
|
}
|
|
});
|
|
mockConsumerDisconnect = jest.fn();
|
|
mockConsumerCreate = jest.fn(() =>
|
|
mock<Consumer>({
|
|
connect: mockConsumerConnect,
|
|
subscribe: mockConsumerSubscribe,
|
|
run: mockConsumerRun,
|
|
disconnect: mockConsumerDisconnect,
|
|
on: jest.fn(() => jest.fn()),
|
|
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',
|
|
},
|
|
}),
|
|
);
|
|
|
|
// Helper to publish a single message using eachBatch (since we now always use eachBatch)
|
|
publishMessage = async (message: Partial<KafkaMessage>) => {
|
|
await mockEachBatchHolder.handler({
|
|
batch: {
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
highWatermark: '100',
|
|
messages: [
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('messageKey'),
|
|
offset: '0',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message'),
|
|
headers: {} as IHeaders,
|
|
...message,
|
|
},
|
|
] as RecordBatchEntry[],
|
|
isEmpty: () => false,
|
|
firstOffset: () => '0',
|
|
lastOffset: () => '0',
|
|
offsetLag: () => '0',
|
|
offsetLagLow: () => '0',
|
|
},
|
|
resolveOffset: jest.fn(),
|
|
heartbeat: jest.fn(),
|
|
commitOffsetsIfNecessary: jest.fn(),
|
|
uncommittedOffsets: jest.fn(),
|
|
isRunning: jest.fn(() => true),
|
|
isStale: jest.fn(() => false),
|
|
pause: jest.fn(),
|
|
});
|
|
};
|
|
|
|
const publishBatch = async (messages: Array<Partial<KafkaMessage>>) => {
|
|
await mockEachBatchHolder.handler({
|
|
batch: {
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
highWatermark: '100',
|
|
messages: messages.map((msg, index) => ({
|
|
attributes: 1,
|
|
key: Buffer.from('messageKey'),
|
|
offset: String(index),
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message'),
|
|
headers: {} as IHeaders,
|
|
...msg,
|
|
})) as RecordBatchEntry[],
|
|
isEmpty: () => false,
|
|
firstOffset: () => '0',
|
|
lastOffset: () => String(messages.length - 1),
|
|
offsetLag: () => '0',
|
|
offsetLagLow: () => '0',
|
|
},
|
|
resolveOffset: jest.fn(),
|
|
heartbeat: jest.fn(),
|
|
commitOffsetsIfNecessary: jest.fn(),
|
|
uncommittedOffsets: jest.fn(),
|
|
isRunning: jest.fn(() => true),
|
|
isStale: jest.fn(() => false),
|
|
pause: jest.fn(),
|
|
});
|
|
};
|
|
|
|
// Expose publishBatch for tests that need it
|
|
(global as any).publishBatch = publishBatch;
|
|
|
|
mockKafka = mock<Kafka>({
|
|
consumer: mockConsumerCreate,
|
|
});
|
|
|
|
mockRegistryDecode = jest.fn().mockResolvedValue({ data: 'decoded-data' });
|
|
mockRegistry = mock<SchemaRegistry>({
|
|
decode: mockRegistryDecode,
|
|
});
|
|
|
|
(Kafka as jest.Mock).mockReturnValue(mockKafka);
|
|
(SchemaRegistry as jest.Mock).mockReturnValue(mockRegistry);
|
|
});
|
|
|
|
it('should connect to Kafka and subscribe to topic', async () => {
|
|
const { close, emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
fromBeginning: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(Kafka).toHaveBeenCalledWith({
|
|
clientId: 'n8n-kafka',
|
|
brokers: ['localhost:9092'],
|
|
ssl: false,
|
|
logLevel: logLevel.ERROR,
|
|
});
|
|
|
|
expect(mockConsumerCreate).toHaveBeenCalledWith({
|
|
groupId: 'test-group',
|
|
maxInFlightRequests: null,
|
|
sessionTimeout: 30000,
|
|
heartbeatInterval: 3000,
|
|
rebalanceTimeout: 600000,
|
|
});
|
|
|
|
expect(mockConsumerConnect).toHaveBeenCalled();
|
|
expect(mockConsumerSubscribe).toHaveBeenCalledWith({
|
|
topic: 'test-topic',
|
|
fromBeginning: true,
|
|
});
|
|
expect(mockConsumerRun).toHaveBeenCalled();
|
|
|
|
await publishMessage({
|
|
value: Buffer.from('message'),
|
|
});
|
|
expect(emit).toHaveBeenCalledWith([[{ json: { message: 'message', topic: 'test-topic' } }]]);
|
|
|
|
await close();
|
|
expect(mockConsumerDisconnect).toHaveBeenCalled();
|
|
});
|
|
|
|
it('should handle authentication when credentials are provided', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: true,
|
|
authentication: true,
|
|
username: 'test-user',
|
|
password: 'test-password',
|
|
saslMechanism: 'plain',
|
|
},
|
|
});
|
|
|
|
expect(Kafka).toHaveBeenCalledWith({
|
|
clientId: 'n8n-kafka',
|
|
brokers: ['localhost:9092'],
|
|
ssl: true,
|
|
logLevel: logLevel.ERROR,
|
|
sasl: {
|
|
username: 'test-user',
|
|
password: 'test-password',
|
|
mechanism: 'plain',
|
|
},
|
|
});
|
|
});
|
|
|
|
it('should throw an error if authentication is enabled but credentials are missing', async () => {
|
|
await expect(
|
|
testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: true,
|
|
},
|
|
}),
|
|
).rejects.toThrow(NodeOperationError);
|
|
});
|
|
|
|
it('should use schema registry when enabled', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: true,
|
|
schemaRegistryUrl: 'http://localhost:8081',
|
|
options: { parallelProcessing: true },
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: Buffer.from('test-message'),
|
|
headers: { 'content-type': Buffer.from('application/json') },
|
|
});
|
|
|
|
expect(SchemaRegistry).toHaveBeenCalledWith({
|
|
host: 'http://localhost:8081',
|
|
});
|
|
expect(mockRegistryDecode).toHaveBeenCalledWith(Buffer.from('test-message'));
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[
|
|
{
|
|
json: {
|
|
message: { data: 'decoded-data' },
|
|
topic: 'test-topic',
|
|
},
|
|
},
|
|
],
|
|
]);
|
|
});
|
|
|
|
it('should parse JSON message when jsonParseMessage is true', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
jsonParseMessage: true,
|
|
parallelProcessing: true,
|
|
onlyMessage: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const jsonData = { foo: 'bar' };
|
|
|
|
await publishMessage({
|
|
value: Buffer.from(JSON.stringify(jsonData)),
|
|
});
|
|
|
|
expect(emit).toHaveBeenCalledWith([[{ json: jsonData }]]);
|
|
});
|
|
|
|
it('should include headers when returnHeaders is true', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
returnHeaders: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: Buffer.from('test-message'),
|
|
headers: {
|
|
'content-type': Buffer.from('application/json'),
|
|
'correlation-id': '123456',
|
|
'with-array-value': ['1', '2', '3'],
|
|
empty: undefined,
|
|
},
|
|
});
|
|
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[
|
|
{
|
|
json: {
|
|
message: 'test-message',
|
|
topic: 'test-topic',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'correlation-id': '123456',
|
|
'with-array-value': '1,2,3',
|
|
empty: '',
|
|
},
|
|
},
|
|
},
|
|
],
|
|
]);
|
|
});
|
|
|
|
it('should handle manual trigger mode', async () => {
|
|
const { emit, manualTriggerFunction } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'manual',
|
|
node: {
|
|
typeVersion: 1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await manualTriggerFunction?.();
|
|
|
|
expect(mockConsumerConnect).toHaveBeenCalledTimes(1);
|
|
expect(mockConsumerSubscribe).toHaveBeenCalledTimes(1);
|
|
expect(mockConsumerRun).toHaveBeenCalledTimes(1);
|
|
|
|
expect(emit).not.toHaveBeenCalled();
|
|
|
|
await publishMessage({ value: Buffer.from('test') });
|
|
|
|
expect(emit).toHaveBeenCalledWith([[{ json: { message: 'test', topic: 'test-topic' } }]]);
|
|
});
|
|
|
|
it('should use immediate emit in manual mode even when resolveOffset is onCompletion (v1.3)', async () => {
|
|
const { emit, manualTriggerFunction } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'manual',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onCompletion',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await manualTriggerFunction?.();
|
|
|
|
await publishMessage({ value: Buffer.from('test-message') });
|
|
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[{ json: { message: 'test-message', topic: 'test-topic' } }],
|
|
]);
|
|
expect(emit.mock.calls[0][2]).toBeUndefined();
|
|
});
|
|
|
|
it('should use immediate emit in manual mode even when parallelProcessing is false (v1.1)', async () => {
|
|
const { emit, manualTriggerFunction } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'manual',
|
|
node: {
|
|
typeVersion: 1.1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
parallelProcessing: false,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await manualTriggerFunction?.();
|
|
|
|
await publishMessage({ value: Buffer.from('test-message') });
|
|
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[{ json: { message: 'test-message', topic: 'test-topic' } }],
|
|
]);
|
|
expect(emit.mock.calls[0][2]).toBeUndefined();
|
|
});
|
|
|
|
it('should handle sequential processing when parallelProcessing is false', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
parallelProcessing: false,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishPromise = publishMessage({
|
|
value: Buffer.from('test-message'),
|
|
});
|
|
|
|
// Wait for the async message handler to execute
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
|
|
const deferredPromise = emit.mock.calls[0][2];
|
|
expect(deferredPromise).toBeDefined();
|
|
|
|
deferredPromise?.resolve(mock());
|
|
await publishPromise;
|
|
});
|
|
|
|
it('should keep binary data when keepBinaryData is enabled in v1.2', async () => {
|
|
const messageBuffer = Buffer.from('binary-avro-data');
|
|
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.2,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
keepBinaryData: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: messageBuffer,
|
|
});
|
|
|
|
// Verify that emit was called with binary data structure
|
|
expect(emit).toHaveBeenCalled();
|
|
const emittedData = emit.mock.calls[0][0][0][0]; // [callIndex][argumentIndex][arrayIndex][itemIndex]
|
|
|
|
// Check the structure has both json and binary properties
|
|
expect(emittedData).toHaveProperty('json');
|
|
expect(emittedData).toHaveProperty('binary');
|
|
// Message is converted to string for the json field
|
|
expect(emittedData.json.message).toBe('binary-avro-data');
|
|
expect(emittedData.json.topic).toBe('test-topic');
|
|
// Raw buffer is preserved in binary data for downstream processing
|
|
expect(emittedData.binary).toHaveProperty('data');
|
|
});
|
|
|
|
it('should not keep binary data in v1.0 and v1.1 even if option is set', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
keepBinaryData: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: Buffer.from('test-message'),
|
|
});
|
|
|
|
// Should emit as string, not binary data
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[
|
|
{
|
|
json: {
|
|
message: 'test-message',
|
|
topic: 'test-topic',
|
|
},
|
|
},
|
|
],
|
|
]);
|
|
});
|
|
|
|
it('should convert to string when keepBinaryData is false in v1.2', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.2,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
keepBinaryData: false,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: Buffer.from('test-message'),
|
|
});
|
|
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[
|
|
{
|
|
json: {
|
|
message: 'test-message',
|
|
topic: 'test-topic',
|
|
},
|
|
},
|
|
],
|
|
]);
|
|
});
|
|
|
|
it('should parse JSON and keep binary data when both options are enabled', async () => {
|
|
const jsonData = { foo: 'bar', nested: { value: 123 } };
|
|
const messageBuffer = Buffer.from(JSON.stringify(jsonData));
|
|
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.2,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
keepBinaryData: true,
|
|
jsonParseMessage: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: messageBuffer,
|
|
});
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const emittedData = emit.mock.calls[0][0][0][0];
|
|
|
|
// Should have parsed JSON in the message field
|
|
expect(emittedData.json.message).toEqual(jsonData);
|
|
expect(emittedData.json.topic).toBe('test-topic');
|
|
// Should also have binary data attached
|
|
expect(emittedData.binary).toHaveProperty('data');
|
|
});
|
|
|
|
it('should decode with schema registry and keep binary data when both are enabled', async () => {
|
|
const messageBuffer = Buffer.from('avro-encoded-data');
|
|
const decodedData = { userId: 123, userName: 'test-user' };
|
|
mockRegistryDecode.mockResolvedValue(decodedData);
|
|
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.2,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: true,
|
|
schemaRegistryUrl: 'http://localhost:8081',
|
|
options: {
|
|
keepBinaryData: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: messageBuffer,
|
|
});
|
|
|
|
// Should call schema registry decode
|
|
expect(mockRegistryDecode).toHaveBeenCalledWith(messageBuffer);
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const emittedData = emit.mock.calls[0][0][0][0];
|
|
|
|
// Should have decoded data in the message field
|
|
expect(emittedData.json.message).toEqual(decodedData);
|
|
expect(emittedData.json.topic).toBe('test-topic');
|
|
// Should also have binary data attached
|
|
expect(emittedData.binary).toHaveProperty('data');
|
|
});
|
|
|
|
it('should work with keepBinaryData and onlyMessage together', async () => {
|
|
const jsonData = { result: 'success', data: [1, 2, 3] };
|
|
const messageBuffer = Buffer.from(JSON.stringify(jsonData));
|
|
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.2,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
keepBinaryData: true,
|
|
jsonParseMessage: true,
|
|
onlyMessage: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: messageBuffer,
|
|
});
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const emittedData = emit.mock.calls[0][0][0][0];
|
|
|
|
// onlyMessage should return just the parsed JSON (not wrapped in message/topic)
|
|
expect(emittedData.json).toEqual(jsonData);
|
|
// Should still have binary data attached
|
|
expect(emittedData.binary).toHaveProperty('data');
|
|
});
|
|
|
|
it('should keep binary data with returnHeaders enabled', async () => {
|
|
const messageBuffer = Buffer.from('test-data');
|
|
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.2,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
keepBinaryData: true,
|
|
returnHeaders: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({
|
|
value: messageBuffer,
|
|
headers: {
|
|
'content-type': Buffer.from('application/octet-stream'),
|
|
'message-id': Buffer.from('msg-123'),
|
|
},
|
|
});
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const emittedData = emit.mock.calls[0][0][0][0];
|
|
|
|
// Should have message and headers (message converted to string)
|
|
expect(emittedData.json.message).toBe('test-data');
|
|
expect(emittedData.json.topic).toBe('test-topic');
|
|
expect(emittedData.json.headers).toEqual({
|
|
'content-type': 'application/octet-stream',
|
|
'message-id': 'msg-123',
|
|
});
|
|
// Raw buffer is preserved in binary data for downstream processing
|
|
expect(emittedData.binary).toHaveProperty('data');
|
|
});
|
|
|
|
it('should use custom rebalanceTimeout when provided', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
rebalanceTimeout: 300000,
|
|
sessionTimeout: 20000,
|
|
heartbeatInterval: 2000,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerCreate).toHaveBeenCalledWith({
|
|
groupId: 'test-group',
|
|
maxInFlightRequests: null,
|
|
sessionTimeout: 20000,
|
|
heartbeatInterval: 2000,
|
|
rebalanceTimeout: 300000,
|
|
});
|
|
});
|
|
|
|
it('should register event listeners on consumer', async () => {
|
|
const mockOn = jest.fn(() => jest.fn());
|
|
mockConsumerCreate.mockReturnValueOnce(
|
|
mock<Consumer>({
|
|
connect: mockConsumerConnect,
|
|
subscribe: mockConsumerSubscribe,
|
|
run: mockConsumerRun,
|
|
disconnect: mockConsumerDisconnect,
|
|
on: mockOn,
|
|
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',
|
|
},
|
|
}),
|
|
);
|
|
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
// Verify all event listeners are registered
|
|
expect(mockOn).toHaveBeenCalledTimes(9);
|
|
expect(mockOn).toHaveBeenCalledWith('consumer.connect', expect.any(Function));
|
|
expect(mockOn).toHaveBeenCalledWith('consumer.group_join', expect.any(Function));
|
|
expect(mockOn).toHaveBeenCalledWith('consumer.network.request_timeout', expect.any(Function));
|
|
expect(mockOn).toHaveBeenCalledWith(
|
|
'consumer.received_unsubscribed_topics',
|
|
expect.any(Function),
|
|
);
|
|
expect(mockOn).toHaveBeenCalledWith('consumer.stop', expect.any(Function));
|
|
expect(mockOn).toHaveBeenCalledWith('consumer.disconnect', expect.any(Function));
|
|
expect(mockOn).toHaveBeenCalledWith('consumer.commit_offsets', expect.any(Function));
|
|
expect(mockOn).toHaveBeenCalledWith('consumer.rebalancing', expect.any(Function));
|
|
expect(mockOn).toHaveBeenCalledWith('consumer.crash', expect.any(Function));
|
|
});
|
|
|
|
it('should clean up event listeners on close', async () => {
|
|
const mockRemoveListener1 = jest.fn();
|
|
const mockRemoveListener2 = jest.fn();
|
|
const mockRemoveListener3 = jest.fn();
|
|
const mockRemoveListener4 = jest.fn();
|
|
const mockRemoveListener5 = jest.fn();
|
|
const mockRemoveListener6 = jest.fn();
|
|
const mockRemoveListener7 = jest.fn();
|
|
const mockRemoveListener8 = jest.fn();
|
|
const mockRemoveListener9 = jest.fn();
|
|
|
|
const mockOn = jest
|
|
.fn()
|
|
.mockReturnValueOnce(mockRemoveListener1)
|
|
.mockReturnValueOnce(mockRemoveListener2)
|
|
.mockReturnValueOnce(mockRemoveListener3)
|
|
.mockReturnValueOnce(mockRemoveListener4)
|
|
.mockReturnValueOnce(mockRemoveListener5)
|
|
.mockReturnValueOnce(mockRemoveListener6)
|
|
.mockReturnValueOnce(mockRemoveListener7)
|
|
.mockReturnValueOnce(mockRemoveListener8)
|
|
.mockReturnValueOnce(mockRemoveListener9);
|
|
|
|
const mockConsumerStop = jest.fn();
|
|
|
|
mockConsumerCreate.mockReturnValueOnce(
|
|
mock<Consumer>({
|
|
connect: mockConsumerConnect,
|
|
subscribe: mockConsumerSubscribe,
|
|
run: mockConsumerRun,
|
|
disconnect: mockConsumerDisconnect,
|
|
stop: mockConsumerStop,
|
|
on: mockOn,
|
|
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 { close } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await close();
|
|
|
|
// Verify all event listener removal functions were called
|
|
expect(mockRemoveListener1).toHaveBeenCalled();
|
|
expect(mockRemoveListener2).toHaveBeenCalled();
|
|
expect(mockRemoveListener3).toHaveBeenCalled();
|
|
expect(mockRemoveListener4).toHaveBeenCalled();
|
|
expect(mockRemoveListener5).toHaveBeenCalled();
|
|
expect(mockRemoveListener6).toHaveBeenCalled();
|
|
expect(mockRemoveListener7).toHaveBeenCalled();
|
|
expect(mockRemoveListener8).toHaveBeenCalled();
|
|
expect(mockRemoveListener9).toHaveBeenCalled();
|
|
|
|
// Verify consumer was stopped and disconnected
|
|
expect(mockConsumerStop).toHaveBeenCalled();
|
|
expect(mockConsumerDisconnect).toHaveBeenCalled();
|
|
});
|
|
|
|
it('should use default values for consumer config when options are not provided', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerCreate).toHaveBeenCalledWith({
|
|
groupId: 'test-group',
|
|
maxInFlightRequests: null,
|
|
sessionTimeout: 30000,
|
|
heartbeatInterval: 3000,
|
|
rebalanceTimeout: 600000,
|
|
});
|
|
});
|
|
|
|
it('should handle batch processing when batchSize > 1', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
batchSize: 3,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishBatch = (global as any).publishBatch;
|
|
await publishBatch([
|
|
{ value: Buffer.from('message1') },
|
|
{ value: Buffer.from('message2') },
|
|
{ value: Buffer.from('message3') },
|
|
]);
|
|
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[
|
|
{ json: { message: 'message1', topic: 'test-topic' } },
|
|
{ json: { message: 'message2', topic: 'test-topic' } },
|
|
{ json: { message: 'message3', topic: 'test-topic' } },
|
|
],
|
|
]);
|
|
});
|
|
|
|
it('should process messages in chunks when batch has more messages than batchSize', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
batchSize: 2,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishBatch = (global as any).publishBatch;
|
|
await publishBatch([
|
|
{ value: Buffer.from('message1') },
|
|
{ value: Buffer.from('message2') },
|
|
{ value: Buffer.from('message3') },
|
|
{ value: Buffer.from('message4') },
|
|
{ value: Buffer.from('message5') },
|
|
]);
|
|
|
|
// Should be called 3 times: [msg1, msg2], [msg3, msg4], [msg5]
|
|
expect(emit).toHaveBeenCalledTimes(3);
|
|
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' } },
|
|
{ json: { message: 'message4', topic: 'test-topic' } },
|
|
],
|
|
]);
|
|
expect(emit).toHaveBeenNthCalledWith(3, [
|
|
[{ json: { message: 'message5', topic: 'test-topic' } }],
|
|
]);
|
|
});
|
|
|
|
it('should use fetchMaxBytes and fetchMinBytes when provided', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
fetchMaxBytes: 2097152,
|
|
fetchMinBytes: 1024,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerCreate).toHaveBeenCalledWith({
|
|
groupId: 'test-group',
|
|
maxInFlightRequests: null,
|
|
sessionTimeout: 30000,
|
|
heartbeatInterval: 3000,
|
|
rebalanceTimeout: 600000,
|
|
maxBytesPerPartition: 2097152,
|
|
minBytes: 1024,
|
|
});
|
|
});
|
|
|
|
it('should use partitionsConsumedConcurrently when provided', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
partitionsConsumedConcurrently: 5,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerRun).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
partitionsConsumedConcurrently: 5,
|
|
}),
|
|
);
|
|
});
|
|
|
|
it('should handle batch processing with sequential processing', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
batchSize: 2,
|
|
parallelProcessing: false,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishBatch = (global as any).publishBatch;
|
|
const publishPromise = publishBatch([
|
|
{ value: Buffer.from('message1') },
|
|
{ value: Buffer.from('message2') },
|
|
]);
|
|
|
|
// Wait for the async batch handler to execute
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
|
|
const deferredPromise = emit.mock.calls[0][2];
|
|
expect(deferredPromise).toBeDefined();
|
|
|
|
deferredPromise?.resolve(mock());
|
|
await publishPromise;
|
|
});
|
|
|
|
it('should handle batch processing with JSON parsing', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
batchSize: 2,
|
|
jsonParseMessage: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishBatch = (global as any).publishBatch;
|
|
await publishBatch([
|
|
{ value: Buffer.from(JSON.stringify({ id: 1, name: 'test1' })) },
|
|
{ value: Buffer.from(JSON.stringify({ id: 2, name: 'test2' })) },
|
|
]);
|
|
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[
|
|
{ json: { message: { id: 1, name: 'test1' }, topic: 'test-topic' } },
|
|
{ json: { message: { id: 2, name: 'test2' }, topic: 'test-topic' } },
|
|
],
|
|
]);
|
|
});
|
|
|
|
it('should handle batch processing with headers', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
batchSize: 2,
|
|
returnHeaders: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishBatch = (global as any).publishBatch;
|
|
await publishBatch([
|
|
{
|
|
value: Buffer.from('message1'),
|
|
headers: { 'content-type': Buffer.from('application/json') },
|
|
},
|
|
{
|
|
value: Buffer.from('message2'),
|
|
headers: { 'content-type': Buffer.from('text/plain') },
|
|
},
|
|
]);
|
|
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[
|
|
{
|
|
json: {
|
|
message: 'message1',
|
|
topic: 'test-topic',
|
|
headers: { 'content-type': 'application/json' },
|
|
},
|
|
},
|
|
{
|
|
json: {
|
|
message: 'message2',
|
|
topic: 'test-topic',
|
|
headers: { 'content-type': 'text/plain' },
|
|
},
|
|
},
|
|
],
|
|
]);
|
|
});
|
|
|
|
it('should keep binary data in batch processing when keepBinaryData is enabled in v1.2', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.2,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
options: {
|
|
batchSize: 2,
|
|
keepBinaryData: true,
|
|
parallelProcessing: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishBatch = (global as any).publishBatch;
|
|
await publishBatch([
|
|
{ value: Buffer.from('binary-data-1') },
|
|
{ value: Buffer.from('binary-data-2') },
|
|
]);
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const emittedData = emit.mock.calls[0][0][0];
|
|
|
|
// Both messages should have binary data
|
|
expect(emittedData[0]).toHaveProperty('json');
|
|
expect(emittedData[0]).toHaveProperty('binary');
|
|
expect(emittedData[0].json.message).toBe('binary-data-1');
|
|
expect(emittedData[0].binary).toHaveProperty('data');
|
|
|
|
expect(emittedData[1]).toHaveProperty('json');
|
|
expect(emittedData[1]).toHaveProperty('binary');
|
|
expect(emittedData[1].json.message).toBe('binary-data-2');
|
|
expect(emittedData[1].binary).toHaveProperty('data');
|
|
});
|
|
|
|
describe('version 1.3', () => {
|
|
it('should use default sessionTimeout and heartbeatInterval', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'immediately',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerCreate).toHaveBeenCalledWith({
|
|
groupId: 'test-group',
|
|
maxInFlightRequests: null,
|
|
sessionTimeout: 30000,
|
|
heartbeatInterval: 10000,
|
|
rebalanceTimeout: 600000,
|
|
});
|
|
});
|
|
|
|
it('should use resolveOffset "immediately" and emit without waiting', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'immediately',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
await publishMessage({ value: Buffer.from('test-message') });
|
|
|
|
expect(emit).toHaveBeenCalledWith([
|
|
[{ json: { message: 'test-message', topic: 'test-topic' } }],
|
|
]);
|
|
expect(emit.mock.calls[0][2]).toBeUndefined();
|
|
});
|
|
|
|
it('should use resolveOffset "onCompletion" and wait for execution', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onCompletion',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishPromise = publishMessage({ value: Buffer.from('test-message') });
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const deferredPromise = emit.mock.calls[0][2];
|
|
expect(deferredPromise).toBeDefined();
|
|
|
|
deferredPromise?.resolve(mock());
|
|
await publishPromise;
|
|
});
|
|
|
|
it('should use resolveOffset "onSuccess" and wait for successful execution', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onSuccess',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishPromise = publishMessage({ value: Buffer.from('test-message') });
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const deferredPromise = emit.mock.calls[0][2];
|
|
expect(deferredPromise).toBeDefined();
|
|
|
|
deferredPromise?.resolve(mock<IRun>({ status: 'success' }));
|
|
await publishPromise;
|
|
});
|
|
|
|
it('should return success false when resolveOffset is "onSuccess" and execution fails', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onSuccess',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishPromise = publishMessage({ value: Buffer.from('test-message') });
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const deferredPromise = emit.mock.calls[0][2];
|
|
expect(deferredPromise).toBeDefined();
|
|
|
|
deferredPromise?.resolve(mock<IRun>({ status: 'error' }));
|
|
// The emitter catches the error, logs it, and returns { success: false }
|
|
// instead of throwing, so the promise resolves (not rejects)
|
|
await publishPromise;
|
|
});
|
|
|
|
it('should use sessionTimeout and heartbeatInterval options when provided', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'immediately',
|
|
options: {
|
|
sessionTimeout: 20000,
|
|
heartbeatInterval: 2000,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerCreate).toHaveBeenCalledWith({
|
|
groupId: 'test-group',
|
|
maxInFlightRequests: null,
|
|
sessionTimeout: 20000,
|
|
heartbeatInterval: 2000,
|
|
rebalanceTimeout: 600000,
|
|
});
|
|
});
|
|
|
|
it('should use resolveOffset "onStatus" and resolve when status matches allowed statuses', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onStatus',
|
|
allowedStatuses: ['success', 'error'],
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishPromise = publishMessage({ value: Buffer.from('test-message') });
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const deferredPromise = emit.mock.calls[0][2];
|
|
expect(deferredPromise).toBeDefined();
|
|
|
|
// Should resolve successfully when status is in allowedStatuses
|
|
deferredPromise?.resolve(mock<IRun>({ status: 'error' }));
|
|
await publishPromise;
|
|
});
|
|
|
|
it('should use resolveOffset "onStatus" and fail when status does not match allowed statuses', async () => {
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onStatus',
|
|
allowedStatuses: ['success'],
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const publishPromise = publishMessage({ value: Buffer.from('test-message') });
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
|
|
expect(emit).toHaveBeenCalled();
|
|
const deferredPromise = emit.mock.calls[0][2];
|
|
expect(deferredPromise).toBeDefined();
|
|
|
|
// Should fail when status is not in allowedStatuses
|
|
deferredPromise?.resolve(mock<IRun>({ status: 'error' }));
|
|
// The emitter catches the error, logs it, and returns { success: false }
|
|
await publishPromise;
|
|
});
|
|
|
|
it('should throw error when resolveOffset is "onStatus" but no statuses are selected', async () => {
|
|
await expect(
|
|
testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onStatus',
|
|
allowedStatuses: [],
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
}),
|
|
).rejects.toThrow(NodeOperationError);
|
|
});
|
|
|
|
it('should stop processing batch when isRunning returns false (consumer stopping)', async () => {
|
|
let eachBatchHandler: EachBatchHandler | undefined;
|
|
mockConsumerRun.mockImplementation(({ eachBatch }: ConsumerRunConfig) => {
|
|
if (eachBatch) {
|
|
eachBatchHandler = eachBatch;
|
|
}
|
|
});
|
|
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'immediately',
|
|
options: {
|
|
batchSize: 1,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const mockResolveOffset = jest.fn();
|
|
const mockHeartbeat = jest.fn();
|
|
|
|
// Simulate consumer stopping after first message
|
|
let messageCount = 0;
|
|
const mockIsRunning = jest.fn(() => {
|
|
messageCount++;
|
|
return messageCount <= 1; // Returns true for first message, false after
|
|
});
|
|
|
|
await eachBatchHandler!({
|
|
batch: {
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
highWatermark: '100',
|
|
messages: [
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key1'),
|
|
offset: '0',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message1'),
|
|
headers: {},
|
|
},
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key2'),
|
|
offset: '1',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message2'),
|
|
headers: {},
|
|
},
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key3'),
|
|
offset: '2',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message3'),
|
|
headers: {},
|
|
},
|
|
] as RecordBatchEntry[],
|
|
isEmpty: () => false,
|
|
firstOffset: () => '0',
|
|
lastOffset: () => '2',
|
|
offsetLag: () => '0',
|
|
offsetLagLow: () => '0',
|
|
},
|
|
resolveOffset: mockResolveOffset,
|
|
heartbeat: mockHeartbeat,
|
|
commitOffsetsIfNecessary: jest.fn(),
|
|
uncommittedOffsets: jest.fn(),
|
|
isRunning: mockIsRunning,
|
|
isStale: jest.fn(() => false),
|
|
pause: jest.fn(),
|
|
});
|
|
|
|
// Should only process first message before isRunning returns false
|
|
expect(emit).toHaveBeenCalledTimes(1);
|
|
expect(mockResolveOffset).toHaveBeenCalledTimes(1);
|
|
expect(mockResolveOffset).toHaveBeenCalledWith('0');
|
|
});
|
|
|
|
it('should stop processing batch when isStale returns true (partition revoked)', async () => {
|
|
let eachBatchHandler: EachBatchHandler | undefined;
|
|
mockConsumerRun.mockImplementation(({ eachBatch }: ConsumerRunConfig) => {
|
|
if (eachBatch) {
|
|
eachBatchHandler = eachBatch;
|
|
}
|
|
});
|
|
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'immediately',
|
|
options: {
|
|
batchSize: 1,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const mockResolveOffset = jest.fn();
|
|
const mockHeartbeat = jest.fn();
|
|
|
|
// Simulate partition becoming stale after first message (rebalance occurred)
|
|
let messageCount = 0;
|
|
const mockIsStale = jest.fn(() => {
|
|
messageCount++;
|
|
return messageCount > 1; // Returns false for first message, true after
|
|
});
|
|
|
|
await eachBatchHandler!({
|
|
batch: {
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
highWatermark: '100',
|
|
messages: [
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key1'),
|
|
offset: '0',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message1'),
|
|
headers: {},
|
|
},
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key2'),
|
|
offset: '1',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message2'),
|
|
headers: {},
|
|
},
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key3'),
|
|
offset: '2',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message3'),
|
|
headers: {},
|
|
},
|
|
] as RecordBatchEntry[],
|
|
isEmpty: () => false,
|
|
firstOffset: () => '0',
|
|
lastOffset: () => '2',
|
|
offsetLag: () => '0',
|
|
offsetLagLow: () => '0',
|
|
},
|
|
resolveOffset: mockResolveOffset,
|
|
heartbeat: mockHeartbeat,
|
|
commitOffsetsIfNecessary: jest.fn(),
|
|
uncommittedOffsets: jest.fn(),
|
|
isRunning: jest.fn(() => true),
|
|
isStale: mockIsStale,
|
|
pause: jest.fn(),
|
|
});
|
|
|
|
// Should only process first message before isStale returns true
|
|
expect(emit).toHaveBeenCalledTimes(1);
|
|
expect(mockResolveOffset).toHaveBeenCalledTimes(1);
|
|
expect(mockResolveOffset).toHaveBeenCalledWith('0');
|
|
});
|
|
|
|
it('should use default auto commit settings (autoCommit true, eachBatchAutoResolve false)', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onSuccess',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerRun).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
autoCommit: true,
|
|
eachBatchAutoResolve: false,
|
|
}),
|
|
);
|
|
});
|
|
|
|
it('should enable eachBatchAutoResolve when option is set', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onCompletion',
|
|
options: {
|
|
eachBatchAutoResolve: true,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerRun).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
autoCommit: true,
|
|
eachBatchAutoResolve: true,
|
|
}),
|
|
);
|
|
});
|
|
|
|
it('should use default auto commit settings with onStatus resolveOffset', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'onStatus',
|
|
allowedStatuses: ['success'],
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerRun).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
autoCommit: true,
|
|
eachBatchAutoResolve: false,
|
|
}),
|
|
);
|
|
});
|
|
|
|
it('should use default auto commit settings with immediately resolveOffset', async () => {
|
|
await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'immediately',
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
expect(mockConsumerRun).toHaveBeenCalledWith(
|
|
expect.objectContaining({
|
|
autoCommit: true,
|
|
eachBatchAutoResolve: false,
|
|
}),
|
|
);
|
|
});
|
|
|
|
it('should process all messages when isRunning and isStale return normal values', async () => {
|
|
let eachBatchHandler: EachBatchHandler | undefined;
|
|
mockConsumerRun.mockImplementation(({ eachBatch }: ConsumerRunConfig) => {
|
|
if (eachBatch) {
|
|
eachBatchHandler = eachBatch;
|
|
}
|
|
});
|
|
|
|
const { emit } = await testTriggerNode(KafkaTrigger, {
|
|
mode: 'trigger',
|
|
node: {
|
|
typeVersion: 1.3,
|
|
parameters: {
|
|
topic: 'test-topic',
|
|
groupId: 'test-group',
|
|
useSchemaRegistry: false,
|
|
resolveOffset: 'immediately',
|
|
options: {
|
|
batchSize: 1,
|
|
},
|
|
},
|
|
},
|
|
credential: {
|
|
brokers: 'localhost:9092',
|
|
clientId: 'n8n-kafka',
|
|
ssl: false,
|
|
authentication: false,
|
|
},
|
|
});
|
|
|
|
const mockResolveOffset = jest.fn();
|
|
const mockHeartbeat = jest.fn();
|
|
|
|
await eachBatchHandler!({
|
|
batch: {
|
|
topic: 'test-topic',
|
|
partition: 0,
|
|
highWatermark: '100',
|
|
messages: [
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key1'),
|
|
offset: '0',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message1'),
|
|
headers: {},
|
|
},
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key2'),
|
|
offset: '1',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message2'),
|
|
headers: {},
|
|
},
|
|
{
|
|
attributes: 1,
|
|
key: Buffer.from('key3'),
|
|
offset: '2',
|
|
timestamp: new Date().toISOString(),
|
|
value: Buffer.from('message3'),
|
|
headers: {},
|
|
},
|
|
] as RecordBatchEntry[],
|
|
isEmpty: () => false,
|
|
firstOffset: () => '0',
|
|
lastOffset: () => '2',
|
|
offsetLag: () => '0',
|
|
offsetLagLow: () => '0',
|
|
},
|
|
resolveOffset: mockResolveOffset,
|
|
heartbeat: mockHeartbeat,
|
|
commitOffsetsIfNecessary: jest.fn(),
|
|
uncommittedOffsets: jest.fn(),
|
|
isRunning: jest.fn(() => true),
|
|
isStale: jest.fn(() => false),
|
|
pause: jest.fn(),
|
|
});
|
|
|
|
// Should process all 3 messages
|
|
expect(emit).toHaveBeenCalledTimes(3);
|
|
expect(mockResolveOffset).toHaveBeenCalledTimes(3);
|
|
expect(mockResolveOffset).toHaveBeenNthCalledWith(1, '0');
|
|
expect(mockResolveOffset).toHaveBeenNthCalledWith(2, '1');
|
|
expect(mockResolveOffset).toHaveBeenNthCalledWith(3, '2');
|
|
});
|
|
});
|
|
});
|