chore: wrap offset_flush return in pin<box<_>> (#4306)

crates/fluvio/src/consumer/stream.rs

@@ -1,3 +1,4 @@
+use std::pin::Pin;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, SystemTime};
@@ -12,14 +13,16 @@ use tracing::{info, warn};
 use super::config::OffsetManagementStrategy;
 use super::{offset::OffsetLocalStore, StreamToServer};
 
+type OffsetFlushFuture<'a> = Pin<Box<dyn Future<Output = Result<(), ErrorCode>> + Send + 'a>>;
+
 /// Extension of [`Stream`] trait with offset management capabilities.
 pub trait ConsumerStream: Stream<Item = Result<Record, ErrorCode>> + Unpin {
     /// Mark the offset of the last yelded record as committed. Depending on [`OffsetManagementStrategy`]
     /// it may require a subsequent `offset_flush()` call to take any effect.
     fn offset_commit(&mut self) -> Result<(), ErrorCode>;
 
     /// Send the committed offset to the server. The method waits for the server's acknowledgment before it finishes.
-    fn offset_flush(&mut self) -> impl Future<Output = Result<(), ErrorCode>> + Send;
+    fn offset_flush(&mut self) -> OffsetFlushFuture<'_>;
 }
 
 pub struct MultiplePartitionConsumerStream<T> {
@@ -117,7 +120,7 @@ where
         self.get_mut().offset_commit()
     }
 
-    fn offset_flush(&mut self) -> impl Future<Output = Result<(), ErrorCode>> + Send {
+    fn offset_flush(&mut self) -> OffsetFlushFuture<'_> {
         self.get_mut().offset_flush()
     }
 }
@@ -129,8 +132,8 @@ impl<T: Stream<Item = Result<Record, ErrorCode>> + Unpin> ConsumerStream
         self.offset_mngt.commit()
     }
 
-    fn offset_flush(&mut self) -> impl Future<Output = Result<(), ErrorCode>> + Send {
-        self.offset_mngt.flush()
+    fn offset_flush(&mut self) -> OffsetFlushFuture<'_> {
+        Box::pin(self.offset_mngt.flush())
     }
 }
 
@@ -144,9 +147,9 @@ impl<T: Stream<Item = Result<Record, ErrorCode>> + Unpin> ConsumerStream
         Ok(())
     }
 
-    fn offset_flush(&mut self) -> impl Future<Output = Result<(), ErrorCode>> + Send {
+    fn offset_flush(&mut self) -> OffsetFlushFuture<'_> {
         let futures: Vec<_> = self.offset_mgnts.iter().map(|p| p.flush()).collect();
-        try_join_all(futures).map(|r| r.map(|_| ()))
+        Box::pin(try_join_all(futures).map(|r| r.map(|_| ())))
     }
 }