try to_owned

src/backends/kafka/mod.rs

@@ -44,8 +44,18 @@ impl<'a> KafkaPayload<'a> {
         }
     }
 }
-impl<'a> Clone for KafkaPayload<'a> {
-    fn clone(&self) -> KafkaPayload<'a> {
+// impl<'a> Clone for KafkaPayload<'a> {
+//     fn clone(&self) -> KafkaPayload<'a> {
+//         match self {
+//             Self::Borrowed(ref msg) => Self::Owned(msg.detach()),
+//             Self::Owned(ref msg) => Self::Owned(msg.clone()),
+//         }
+//     }
+// }
+
+impl ToOwned for KafkaPayload<'static> {
+    type Owned = KafkaPayload<'static>;
+    fn to_owned(&self) -> KafkaPayload<'static> {
         match self {
             Self::Borrowed(ref msg) => Self::Owned(msg.detach()),
             Self::Owned(ref msg) => Self::Owned(msg.clone()),

src/processing/mod.rs

@@ -112,12 +112,20 @@ impl<'a> StreamProcessor<'a> {
         } else {
             // Otherwise, we need to try fetch a new message from the consumer,
             // even if there is no active assignment and/or processing strategy.
-            let msg = self.consumer.poll(Some(Duration::from_secs(1)));
+            let msg = self
+                .consumer
+                .poll(Some(Duration::from_secs(1)))
+                .unwrap()
+                .unwrap()
+                .to_owned();
+
+            self.message = Some(msg)
+
             //TODO: Support errors properly
-            match msg {
-                Ok(m) => self.message = m,
-                Err(_) => return Err(RunError::PollError),
-            }
+            // match msg {
+            //     Ok(m) => self.message = m,
+            //     Err(_) => return Err(RunError::PollError),
+            // }
         }
 
         let mut trait_callbacks = self.strategies.lock().unwrap();