fix server -> kafka connection

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

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
 }

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);