mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2024-11-13 16:14:07 -08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

feat(Kafka Trigger Node): Add additional options (#3600)

Browse source 
* 🔨 additional options to kafka trigger

* ⚡ option for maxInFlightRequests

* ⚡ Small change

Co-authored-by: ricardo <ricardoespinoza105@gmail.com>
This commit is contained in:
Michael Kret 2022-07-27 18:00:39 +03:00 committed by GitHub
parent 3ebfa45570
commit 3496a39788
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 46 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts
View file

@ -94,6 +94,36 @@ export class KafkaTrigger implements INodeType {
default: false, default: false,
description: 'Whether to allow sending message to a previously non exisiting topic', description: 'Whether to allow sending message to a previously non exisiting topic',
}, },
{
displayName: 'Auto Commit Threshold',
name: 'autoCommitThreshold',
type: 'number',
default: 0,
description: 'The consumer will commit offsets after resolving a given number of messages',
},
{
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: 'Heartbeat Interval',
name: 'heartbeatInterval',
type: 'number',
default: 3000,
description: 'Heartbeats are used to ensure that the consumer\'s session stays active',
hint: 'The value must be set lower than Session Timeout',
},
{
displayName: 'Max Number of Requests',
name: 'maxInFlightRequests',
type: 'number',
default: 0,
description: 'Max number of requests that may be in progress at any time. If falsey then no limit.',
},
{ {
displayName: 'Read Messages From Beginning', displayName: 'Read Messages From Beginning',
name: 'fromBeginning', name: 'fromBeginning',
@ -122,13 +152,6 @@ export class KafkaTrigger implements INodeType {
default: false, default: false,
description: 'Whether to return only the message property', description: 'Whether to return only the message property',
}, },
{
displayName: 'Session Timeout',
name: 'sessionTimeout',
type: 'number',
default: 30000,
description: 'The time to await a response in ms',
},
{ {
displayName: 'Return Headers', displayName: 'Return Headers',
name: 'returnHeaders', name: 'returnHeaders',
@ -136,6 +159,14 @@ export class KafkaTrigger implements INodeType {
default: false, default: false,
description: 'Whether to return the headers received from Kafka', description: 'Whether to return the headers received from Kafka',
}, },
{
displayName: 'Session Timeout',
name: 'sessionTimeout',
type: 'number',
default: 30000,
description: 'The time to await a response in ms',
hint: 'Value in milliseconds',
},
], ],
}, },
], ],
@ -175,7 +206,12 @@ export class KafkaTrigger implements INodeType {
const kafka = new apacheKafka(config); const kafka = new apacheKafka(config);
const consumer = kafka.consumer({ groupId }); const consumer = kafka.consumer({
groupId,
maxInFlightRequests: this.getNodeParameter('options.maxInFlightRequests', 0) as number,
sessionTimeout: this.getNodeParameter('options.sessionTimeout', 30000) as number,
heartbeatInterval: this.getNodeParameter('options.heartbeatInterval', 3000) as number,
});
await consumer.connect(); await consumer.connect();
@ -191,6 +227,8 @@ export class KafkaTrigger implements INodeType {
const startConsumer = async () => { const startConsumer = async () => {
await consumer.run({ await consumer.run({
autoCommitInterval: options.autoCommitInterval as number || null,
autoCommitThreshold: options.autoCommitThreshold as number || null,
eachMessage: async ({ topic, message }) => { eachMessage: async ({ topic, message }) => {
let data: IDataObject = {}; let data: IDataObject = {};