diff --git a/pong/src/event.rs b/pong/src/event.rs
index 0c4917e..a40981a 100644
--- a/pong/src/event.rs
+++ b/pong/src/event.rs
@@ -13,7 +13,7 @@ pub mod event {
     }
 
     #[async_trait]
-    pub trait EventWriterImpl {
+    pub trait EventWriterImpl : Send {
         async fn write(&mut self, events: Vec<EventWrapper>) -> Result<(), String>;
     }
 
@@ -82,7 +82,7 @@ pub mod event {
     }
 
     #[async_trait]
-    pub trait EventReaderImpl {
+    pub trait EventReaderImpl : Send {
         async fn read(&mut self) -> Result<Vec<EventWrapper>, String>;
     }
 
diff --git a/server/src/session_manager.rs b/server/src/session_manager.rs
index 600919e..b7a9a30 100644
--- a/server/src/session_manager.rs
+++ b/server/src/session_manager.rs
@@ -215,7 +215,7 @@ impl SessionManager {
         return Ok((reader.unwrap(), writer.unwrap()));
     }
 
-    pub fn get_session_reader(
+    pub async fn get_session_reader(
         &self,
         session_id: &str,
         topics: &[&str],
@@ -225,7 +225,7 @@ impl SessionManager {
             return Err(format!("Unable to find session with hash {}", session_id));
         }
         let session = session.unwrap();
-        let kafka_reader = KafkaSessionEventReaderImpl::new(&self.kafka_host, &session, topics);
+        let kafka_reader = KafkaSessionEventReaderImpl::new(&self.kafka_host, &session, topics).await;
         if let Err(_) = kafka_reader {
             return Err("Unable to create kafka reader.".to_string());
         }