Repository: n8n-io/n8n
File: packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts (TypeScript)

Code before:

		const kafka = new apacheKafka(config);
		const consumer = kafka.consumer({
			groupId,
			maxInFlightRequests,
			sessionTimeout: this.getNodeParameter('options.sessionTimeout', 30000) as number,
			heartbeatInterval: this.getNodeParameter('options.heartbeatInterval', 3000) as number,
		});

		// The "closeFunction" function gets called by n8n whenever
		// the workflow gets deactivated and can so clean up.
		async function closeFunction() {
			await consumer.disconnect();
		}

		const startConsumer = async () => {
			await consumer.connect();

			await consumer.subscribe({ topic, fromBeginning: options.fromBeginning ? true : false });
			await consumer.run({
				autoCommitInterval: (options.autoCommitInterval as number) || null,
				autoCommitThreshold: (options.autoCommitThreshold as number) || null,
				eachMessage: async ({ topic: messageTopic, message }) => {
					let data: IDataObject = {};
					let value: any;
					let binary: { [key: string]: any } = {};

					// Always perform parsing/decoding flow
					value = message.value?.toString() as string;

					if (options.jsonParseMessage) {
						try {
							value = JSON.parse(value);
						} catch (error) {}
					}

					if (useSchemaRegistry) {
						try {
							const registry = new SchemaRegistry({ host: schemaRegistryUrl });
							value = await registry.decode(message.value as Buffer);
						} catch (error) {}
					}

					// Additionally, preserve raw binary data for downstream processing (only in v1.2+)
					if (nodeVersion >= 1.2 && options.keepBinaryData && message.value) {
						const binaryData = await this.helpers.prepareBinaryData(
							message.value as Buffer,
							'message',
							'application/octet-stream',
						);
						binary.data = binaryData;

Review comment (suggestion):
Same here:

```suggestion
					let binary: IBinaryKeyData = {};
```

Code after:
						type: 'boolean',
						default: false,
						description: 'Whether to return the headers received from Kafka',
					},
					{
						displayName: 'Rebalance Timeout',
						name: 'rebalanceTimeout',
						type: 'number',
						default: 600000,
						description: 'The maximum time allowed for a consumer to join the group',
					},
					{
						displayName: 'Session Timeout',
						name: 'sessionTimeout',
						type: 'number',
						default: 30000,
						description: 'The time to await a response in ms',
						hint: 'Value in milliseconds',
					},
				],
			},
		],
	};

	async trigger(this: ITriggerFunctions): Promise<ITriggerResponse> {
		const topic = this.getNodeParameter('topic') as string;

		const groupId = this.getNodeParameter('groupId') as string;

		const credentials = await this.getCredentials('kafka');

		const brokers = ((credentials.brokers as string) ?? '').split(',').map((item) => item.trim());

		const clientId = credentials.clientId as string;

		const ssl = credentials.ssl as boolean;

		const options = this.getNodeParameter('options', {}) as KafkaTriggerOptions;

		const nodeVersion = this.getNode().typeVersion;
		options.nodeVersion = nodeVersion;

		const config: KafkaConfig = {
			clientId,
			brokers,
			ssl,
			logLevel: logLevel.ERROR,
		};

		if (credentials.authentication === true) {
			if (!(credentials.username && credentials.password)) {