fix: disallow now_or_never (and async read)

clippy.toml

@@ -1,3 +1,5 @@
 disallowed-methods = [
+    { path = "tokio::io::AsyncReadExt::read", reason = "Use `read_exact`, because you likely forgot to use the IO amount. Also see `unused_io_amount` clippy lint." },
     { path = "tokio::io::AsyncWriteExt::write", reason = "Use `write_all`, because you likely forgot to use the IO amount. Also see `unused_io_amount` clippy lint." },
+    { path = "futures_util::future::future::FutureExt::now_or_never", reason = "Tokio might preempt tasks at any time, see https://tokio.rs/blog/2020-04-preemption" }
 ]

src/client/consumer.rs

@@ -450,12 +450,20 @@ mod tests {
         assert_eq!(high_watermark, 5);
 
         let (record_and_offset, high_watermark) =
-            stream.next().now_or_never().unwrap().unwrap().unwrap();
+            tokio::time::timeout(Duration::from_millis(1), stream.next())
+                .await
+                .unwrap()
+                .unwrap()
+                .unwrap();
         assert_eq!(record_and_offset.offset, 4);
         assert_eq!(high_watermark, 5);
 
         let (record_and_offset, high_watermark) =
-            stream.next().now_or_never().unwrap().unwrap().unwrap();
+            tokio::time::timeout(Duration::from_millis(1), stream.next())
+                .await
+                .unwrap()
+                .unwrap()
+                .unwrap();
         assert_eq!(record_and_offset.offset, 5);
         assert_eq!(high_watermark, 5);
 

src/client/producer.rs

@@ -439,7 +439,10 @@ where
         // Flush data
         Self::flush_impl(&mut inner, self.client.as_ref(), self.compression).await;
 
-        extract(&result_slot.now_or_never().expect("just flushed"), tag)
+        // We need to poll the result slot here to make the result available via `peek`, otherwise the next thread in
+        // this critical section will flush the producer a 2nd time. We are using an ordinary `.await` call here instead
+        // of `now_or_never` because tokio might preempt us (or to be precise: might preempt polling from the result slot).
+        extract(&result_slot.await, tag)
     }
 
     /// Flushed out data from aggregator.

src/client/producer/broadcast.rs

@@ -69,6 +69,8 @@ impl<T: Clone> std::future::Future for ReceiveFut<T> {
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
     use super::*;
 
     #[tokio::test]
@@ -93,10 +95,21 @@ mod tests {
 
         // Test drop
         let broadcast: BroadcastOnce<usize> = Default::default();
-        let r1 = broadcast.receive();
+        let mut r1 = broadcast.receive();
         let r2 = broadcast.receive();
-        assert!(r1.now_or_never().is_none());
+        assert_is_pending(&mut r1).await;
         broadcast.broadcast(5);
         assert_eq!(r2.await, 5);
     }
+
+    async fn assert_is_pending<F, T>(f: &mut F)
+    where
+        F: Future<Output = T> + Send + Unpin,
+        T: std::fmt::Debug,
+    {
+        tokio::select! {
+            x = f => panic!("future is not pending, yielded: {x:?}"),
+            _ = tokio::time::sleep(Duration::from_millis(1)) => {},
+        };
+    }
 }