changed kafka and zookeper deployments

Author: Javedmv

manifest/deployments/kafka.yaml

@@ -15,38 +15,28 @@ spec:
         app: kafka
     spec:
       containers:
-      - name: kafka
-        image: bitnami/kafka:latest
-        ports:
-        - containerPort: 9092
-          name: plaintext
-        - containerPort: 9093
-          name: controller
-        env:
-        - name: ALLOW_PLAINTEXT_LISTENER
-          value: "yes"
-        - name: KAFKA_CFG_BROKER_ID
-          value: "1"
-        - name: KAFKA_CFG_NODE_ID
-          value: "1"  
-        - name: KAFKA_CFG_PROCESS_ROLES
-          value: "broker,controller"
-        - name: KAFKA_CFG_CONTROLLER_QUORUM_VOTERS
-          value: "1@kafka:9093"
-        - name: KAFKA_CFG_LISTENERS
-          value: "PLAINTEXT://:9092,CONTROLLER://:9093"
-        - name: KAFKA_CFG_ADVERTISED_LISTENERS
-          value: "PLAINTEXT://kafka:9092"
-        - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
-          value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
-        - name: KAFKA_CFG_CONTROLLER_LISTENER_NAMES
-          value: "CONTROLLER"
-        - name: KAFKA_CFG_INTER_BROKER_LISTENER_NAME
-          value: "PLAINTEXT"
-        - name: KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR
-          value: "1"
-        - name: KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE
-          value: "true"
+        - name: kafka
+          image: bitnami/kafka:latest
+          ports:
+            - containerPort: 9092
+              name: plaintext
+          env:
+            - name: KAFKA_CFG_ZOOKEEPER_CONNECT
+              value: "zookeeper:2181"
+            - name: ALLOW_PLAINTEXT_LISTENER
+              value: "yes"
+            - name: KAFKA_CFG_LISTENERS
+              value: "PLAINTEXT://:9092"
+            - name: KAFKA_CFG_ADVERTISED_LISTENERS
+              value: "PLAINTEXT://kafka:9092"
+            - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
+              value: "PLAINTEXT:PLAINTEXT"
+            - name: KAFKA_CFG_INTER_BROKER_LISTENER_NAME
+              value: "PLAINTEXT"
+            - name: KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR
+              value: "1"
+            - name: KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE
+              value: "true"
 ---
 apiVersion: v1
 kind: Service
@@ -56,11 +46,8 @@ metadata:
     app: kafka
 spec:
   ports:
-  - port: 9092
-    targetPort: 9092
-    name: plaintext
-  - port: 9093
-    targetPort: 9093
-    name: controller
+    - port: 9092
+      targetPort: 9092
+      name: plaintext
   selector:
-    app: kafka
+    app: kafka

manifest/deployments/zookeeper.yaml

@@ -15,14 +15,27 @@ spec:
         app: zookeeper
     spec:
       containers:
-      - name: zookeeper
-        image: bitnami/zookeeper:latest
-        ports:
-        - containerPort: 2181
-        env:
-        - name: ZOOKEEPER_CLIENT_PORT
-          value: "2181"
-        - name: ZOOKEEPER_TICK_TIME
-          value: "2000"
-        - name: ALLOW_ANONYMOUS_LOGIN
-          value: "yes"  # This line enables anonymous login
+        - name: zookeeper
+          image: bitnami/zookeeper:latest
+          ports:
+            - containerPort: 2181
+          env:
+            - name: ZOOKEEPER_CLIENT_PORT
+              value: "2181"
+            - name: ZOOKEEPER_TICK_TIME
+              value: "2000"
+            - name: ALLOW_ANONYMOUS_LOGIN
+              value: "yes"
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: zookeeper
+  labels:
+    app: zookeeper
+spec:
+  ports:
+    - port: 2181
+      targetPort: 2181
+  selector:
+    app: zookeeper

payment-srv/src/application/useCases/enrollStudentInCourseUseCase.ts

@@ -1,3 +1,4 @@
+import { producer } from "../../infrastructure/kafka";
 import { sendEnrollPaidUserProducer } from "../../infrastructure/kafka/producers";
 import { IDependencies } from "../interfaces/IDependencies";
 

payment-srv/src/index.ts

@@ -1,12 +1,13 @@
 import Server from './presentation/server'
 import { connectDB } from './__boot/config';
 import { runConsumer, stopConsumer } from './__boot/consumer';
+import { producer } from './infrastructure/kafka';
 
 (async () => {
     try {
         Server;
 
-        await Promise.all([connectDB(), runConsumer()])
+        await Promise.all([connectDB(),producer.connect(), runConsumer()])
         // TODO: disconnected the consumer reconnect it when needed
         .then(() => {
             console.log("kafka consumer is running")
@@ -37,8 +38,12 @@ import { runConsumer, stopConsumer } from './__boot/consumer';
             console.log("Stopping Kafka consumer...");
             await stopConsumer();
             console.log("Kafka consumer stopped.");
+    
+            console.log("Disconnecting Kafka producer...");
+            await producer.disconnect();
+            console.log("Kafka producer disconnected.");
         } catch (error) {
-            console.error("Error while stopping Kafka consumer:", error);
+            console.error("Error while shutting down services:", error);
         } finally {
             console.log("Process exiting...");
             process.exit(0);