fix: auto management odd pauses in consumer offset (#4504)

* fix: auto management odd pauses in consumer offset

* feat: add flusher_check_period in consumer builder

Author: fraidev

crates/fluvio-test/src/tests/consumer_offsets/auto/flush.rs

@@ -43,11 +43,13 @@ pub async fn test_strategy_auto_periodic_flush(
     for _ in 0..10 {
         ensure_read(&mut stream, &mut counts).await?;
     }
-    sleep(Duration::from_secs(2)).await; // yeild  to drive auto flush flow
+    sleep(Duration::from_secs(3)).await; // yeild  to drive auto flush flow
 
-    let consumer = find_consumer(client, &consumer_id, 0).await?;
-    ensure!(consumer.is_some());
-    ensure!(consumer.unwrap().offset > 0i64);
+    for (partition, offset) in counts.iter().enumerate().take(partitions) {
+        let consumer = find_consumer(client, &consumer_id, partition).await?;
+        ensure!(consumer.is_some());
+        ensure!(consumer.unwrap().offset == *offset);
+    }
 
     drop(stream); //we keep the stream alive to prevent flush on drop occuring
 

crates/fluvio/src/consumer/config.rs

@@ -11,6 +11,7 @@ use crate::{FluvioError, Offset};
 use super::MAX_FETCH_BYTES;
 
 const DEFAULT_OFFSET_FLUSH_PERIOD: Duration = Duration::from_secs(10);
+const DEFAULT_OFFSET_FLUSHER_CHECK_PERIOD: Duration = Duration::from_millis(100);
 const DEFAULT_RETRY_MODE: RetryMode = RetryMode::TryUntil(100);
 
 /// Configures the behavior of consumer fetching and streaming
@@ -93,6 +94,8 @@ pub struct ConsumerConfigExt {
     pub offset_strategy: OffsetManagementStrategy,
     #[builder(default = "DEFAULT_OFFSET_FLUSH_PERIOD")]
     pub offset_flush: Duration,
+    #[builder(default = "DEFAULT_OFFSET_FLUSHER_CHECK_PERIOD")]
+    pub offset_flusher_check_period: Duration,
     #[builder(default)]
     pub disable_continuous: bool,
     #[builder(default = "*MAX_FETCH_BYTES")]
@@ -118,6 +121,7 @@ impl ConsumerConfigExt {
         Option<String>,
         OffsetManagementStrategy,
         Duration,
+        Duration,
     ) {
         let Self {
             topic: _,
@@ -131,6 +135,7 @@ impl ConsumerConfigExt {
             smartmodule,
             offset_strategy,
             offset_flush,
+            offset_flusher_check_period,
             retry_mode: _,
         } = self;
 
@@ -147,6 +152,7 @@ impl ConsumerConfigExt {
             offset_consumer,
             offset_strategy,
             offset_flush,
+            offset_flusher_check_period,
         )
     }
 }
@@ -185,6 +191,7 @@ impl From<ConsumerConfigExt> for ConsumerConfig {
             offset_start: _,
             offset_strategy: _,
             offset_flush: _,
+            offset_flusher_check_period: _,
             disable_continuous,
             max_bytes,
             isolation,

crates/fluvio/src/consumer/mod.rs

@@ -73,30 +73,18 @@ pub type BoxConsumerStream =
 pub type BoxConsumerStream =
     Pin<Box<dyn ConsumerStream<Item = Result<ConsumerRecord, ErrorCode>> + Send + 'static>>;
 
+type ShararedConsumerStream = Arc<Mutex<BoxConsumerStream>>;
+
+type ConsumerFutureOutput = (
+    ShararedConsumerStream,
+    Option<Result<(ConsumerRecord, Option<i64>), ErrorCode>>,
+);
+
 /// Type alias to access consume stream as a future.
 #[cfg(target_arch = "wasm32")]
-type BoxConsumerFuture = Pin<
-    Box<
-        dyn Future<
-                Output = (
-                    Arc<Mutex<BoxConsumerStream>>,
-                    Option<Result<(ConsumerRecord, Option<i64>), ErrorCode>>,
-                ),
-            > + 'static,
-    >,
->;
+type BoxConsumerFuture = Pin<Box<dyn Future<Output = ConsumerFutureOutput> + 'static>>;
 #[cfg(not(target_arch = "wasm32"))]
-type BoxConsumerFuture = Pin<
-    Box<
-        dyn Future<
-                Output = (
-                    Arc<Mutex<BoxConsumerStream>>,
-                    Option<Result<(ConsumerRecord, Option<i64>), ErrorCode>>,
-                ),
-            > + Send
-            + 'static,
-    >,
->;
+type BoxConsumerFuture = Pin<Box<dyn Future<Output = ConsumerFutureOutput> + Send + 'static>>;
 
 /// An interface for consuming events from a particular partition
 ///
@@ -617,7 +605,8 @@ where
         &self,
         config: ConsumerConfigExt,
     ) -> Result<SinglePartitionConsumerStream<impl Stream<Item = Result<Record, ErrorCode>>>> {
-        let (offset, config, consumer_id, strategy, flush_period) = config.into_parts();
+        let (offset, config, consumer_id, strategy, flush_period, flusher_check_period) =
+            config.into_parts();
         let (stream, start_offset, stream_to_server) = self
             .inner_stream_batches_with_config(offset, config, consumer_id)
             .await?;
@@ -642,6 +631,7 @@ where
             flattened,
             strategy,
             flush_period,
+            flusher_check_period,
             stream_to_server,
         ))
     }