From 99f6ccaf0355141b17c9da3df4fa1e6f42648583 Mon Sep 17 00:00:00 2001
From: Thilo Behnke <thilo.behnke@gmx.net>
Date: Wed, 10 Aug 2022 15:27:56 +0200
Subject: [PATCH] fix server -> kafka connection

---
 k8s.yml                       |  5 +++--
 server/src/kafka.rs           | 14 +++++++++++---
 server/src/session_manager.rs |  1 -
 3 files changed, 14 insertions(+), 6 deletions(-)

diff --git a/k8s.yml b/k8s.yml
index cc8e76f..22402ed 100644
--- a/k8s.yml
+++ b/k8s.yml
@@ -73,9 +73,10 @@ spec:
         - name: KAFKA_CFG_INTER_BROKER_LISTENER_NAME
           value: K8S
         - name: KAFKA_CFG_LISTENERS
-          value: LOCAL://:9093,K8S://kafka:9092
+          # TODO: Does not work with minikube
+          value: LOCAL://:9093,K8S://0.0.0.0:9092
         - name: KAFKA_CFG_ADVERTISED_LISTENERS
-          value: LOCAL://127.0.0.1:9093,K8S://kafka:9092
+          value: LOCAL://127.0.0.1:9093,K8S://kafka-service:9092
         - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
           value: LOCAL:PLAINTEXT,K8S:PLAINTEXT
         - name: KAFKA_CFG_ZOOKEEPER_CONNECT
diff --git a/server/src/kafka.rs b/server/src/kafka.rs
index 91a7947..02effbc 100644
--- a/server/src/kafka.rs
+++ b/server/src/kafka.rs
@@ -38,6 +38,7 @@ impl KafkaSessionEventWriterImpl {
             }
             let producer = producer.unwrap();
             producers.insert(topic.to_owned(), producer);
+            debug!("Successfully connected kafka producer for topic={}, partition={}", topic, partition)
         }
         KafkaSessionEventWriterImpl { topics: owned_topics, partitions: vec![partition.clone()], producers }
     }
@@ -59,7 +60,7 @@ impl EventWriterImpl for KafkaSessionEventWriterImpl {
             }
         }
         for topic_events in by_topic {
-            let mut producer = self.producers.get_mut(&topic_events.0);
+            let producer = self.producers.get_mut(&topic_events.0);
             if let None = producer {
                 let available = self.producers.keys().collect::<Vec<&String>>();
                 return Err(format!("Could not find producer for topic: {}. Available topic producers: {:?}", &topic_events.0, available));
@@ -97,9 +98,16 @@ async fn write_events(events: Vec<EventWrapper>, producer: &mut PartitionClient)
         }
     }
 
+    debug!("producer ready to write records to kafka: {:?}", records);
     let res = match producer.produce(records, Compression::NoCompression).await {
-        Ok(_) => Ok(()),
-        Err(e) => Err(format!("{:?}", e)),
+        Ok(_) => {
+            debug!("producer successfully wrote records to kafka");
+            Ok(())
+        },
+        Err(e) => {
+            debug!("producer failed to write records to kafka: {:?}", e);
+            Err(format!("{:?}", e))
+        },
     };
     res
 }
diff --git a/server/src/session_manager.rs b/server/src/session_manager.rs
index 8081384..371778e 100644
--- a/server/src/session_manager.rs
+++ b/server/src/session_manager.rs
@@ -179,7 +179,6 @@ impl SessionManager {
         let json_event = json_event.unwrap();
         info!("preparing to write session event to kafka: {}", json_event);
         let mut session_writer = self.get_session_writer(session_id).await.unwrap();
-        // TODO: This gets stuck in K8S - fix
         let session_event_write = session_writer.write_to_session("session", vec![&json_event]).await;
         if let Err(e) = session_event_write {
             let message = format!("Failed to write session event to kafka: {:?}", e);