wip

Author: untitaker

src/backends/kafka/mod.rs

@@ -180,7 +180,7 @@ impl ArroyoConsumer for KafkaConsumer {
     }
 
     fn poll(
-        &self,
+        &mut self,
         timeout: Option<Duration>,
     ) -> Result<Option<ArroyoMessage<Payload<'_>>>, PollError> {
         let duration = timeout.unwrap_or(Duration::from_millis(100));

src/backends/local/mod.rs

@@ -130,7 +130,7 @@ impl Consumer for LocalConsumer {
         Ok(())
     }
 
-    fn poll(&self, _timeout: Option<Duration>) -> Result<Option<Message<Payload>>, PollError> {
+    fn poll(&mut self, _timeout: Option<Duration>) -> Result<Option<Message<Payload>>, PollError> {
         if self.closed {
             return Err(PollError::ConsumerClosed);
         }

src/backends/mod.rs

@@ -109,7 +109,7 @@ pub trait Consumer {
     /// consumer attempts to read from an invalid location in one of it's
     /// assigned partitions. (Additional details can be found in the
     /// docstring for ``Consumer.seek``.)
-    fn poll(&self, timeout: Option<Duration>) -> Result<Option<Message<Payload>>, PollError>;
+    fn poll(&mut self, timeout: Option<Duration>) -> Result<Option<Message<Payload>>, PollError>;
 
     /// Pause consuming from the provided partitions.
     ///