update kafka client to take in environement variables to declare SASL/SSL options

This commit is contained in:
Francis Cao 2024-12-14 11:49:15 -08:00
parent d43293da72
commit 8f692a150c
2 changed files with 14 additions and 5 deletions

View file

@ -23,5 +23,12 @@ if (!process.env.SKIP_DB_CHECK && !process.env.DATABASE_TYPE) {
}
if (process.env.CLOUD_MODE) {
checkMissing(['CLOUD_URL', 'KAFKA_BROKER', 'KAFKA_URL', 'REDIS_URL']);
checkMissing([
'CLOUD_URL',
'KAFKA_BROKER',
'KAFKA_URL',
'REDIS_URL',
'KAFKA_SASL_MECHANISM',
'KAFKA_SSL',
]);
}

View file

@ -1,5 +1,5 @@
import debug from 'debug';
import { Kafka, Mechanism, Producer, RecordMetadata, SASLOptions, logLevel } from 'kafkajs';
import { Kafka, Producer, RecordMetadata, SASLOptions, logLevel } from 'kafkajs';
import { KAFKA, KAFKA_PRODUCER } from 'lib/db';
import * as tls from 'tls';
@ -12,13 +12,15 @@ const enabled = Boolean(process.env.KAFKA_URL && process.env.KAFKA_BROKER);
function getClient() {
const { username, password } = new URL(process.env.KAFKA_URL);
const brokers = process.env.KAFKA_BROKER.split(',');
const sslEnabled = process.env.KAFKA_SSL.toLowerCase() === 'true';
const mechanism = process.env.KAFKA_SASL_MECHANISM as 'plain' | 'scram-sha-256' | 'scram-sha-512';
const ssl: { ssl?: tls.ConnectionOptions | boolean; sasl?: SASLOptions | Mechanism } =
const ssl: { ssl?: tls.ConnectionOptions | boolean; sasl?: SASLOptions } =
username && password
? {
ssl: true,
ssl: sslEnabled,
sasl: {
mechanism: 'scram-sha-256',
mechanism,
username,
password,
},