fix: add SASL

Author: izadoesdev

apps/basket/src/lib/kafka.test.ts

@@ -20,6 +20,11 @@ describe('Kafka Producer and Consumer', () => {
 				initialRetryTime: 100,
 				retries: 8,
 			},
+			sasl: {
+				mechanism: 'scram-sha-256',
+				username: process.env.KAFKA_USER as string,
+				password: process.env.KAFKA_PASSWORD as string,
+			},
 		});
 
 		producer = kafka.producer();
@@ -72,6 +77,7 @@ describe('Kafka Producer and Consumer', () => {
 			await admin.connect();
 
 			const topics = await admin.listTopics();
+			console.log('topics', topics);
 			expect(Array.isArray(topics)).toBe(true);
 
 			await admin.disconnect();

apps/basket/src/lib/producer.test.ts

@@ -16,6 +16,11 @@ describe('Producer Module', () => {
 		kafka = new Kafka({
 			clientId: 'producer-test',
 			brokers: [KAFKA_BROKERS],
+			sasl: {
+				mechanism: 'scram-sha-256',
+				username: process.env.KAFKA_USER as string,
+				password: process.env.KAFKA_PASSWORD as string,
+			},
 			retry: {
 				initialRetryTime: 100,
 				retries: 8,
@@ -264,9 +269,6 @@ describe('Producer Module', () => {
 			expect(disconnectProducer).toBeDefined();
 		});
 
-		it('should be idempotent', async () => {
-			expect(disconnectProducer).toBeDefined();
-		});
 	});
 
 	describe('Analytics Events', () => {

apps/basket/src/lib/producer.ts

@@ -68,7 +68,7 @@ if (BROKER) {
 		requestTimeout: KAFKA_TIMEOUT,
 		sasl:{
 			mechanism: 'scram-sha-256',
-			username: process.env.KAFKA_USERNAME as string,
+			username: process.env.KAFKA_USER as string,
 			password: process.env.KAFKA_PASSWORD as string,
 		}
 	});

infra/ingest/vector.yaml

@@ -2,6 +2,10 @@ sources:
   redpanda:
     type: kafka
     bootstrap_servers: "${REDPANDA_BROKER}"
+    sasl:
+      mechanism: "scram-sha-256"
+      username: "${VECTOR_KAFKA_USER}"
+      password: "${VECTOR_KAFKA_PASSWORD}"
     group_id: vector-analytics
     topics:
       - analytics-events