test: extend consumer stream tests

Author: crepererum

tests/client.rs

@@ -1,7 +1,7 @@
 use assert_matches::assert_matches;
 use rskafka::{
     client::{
-        error::{Error as ClientError, ProtocolError, RequestError},
+        error::{Error as ClientError, ProtocolError},
         partition::{Compression, OffsetAt},
         ClientBuilder,
     },
@@ -10,9 +10,7 @@ use rskafka::{
 use std::{collections::BTreeMap, str::FromStr, sync::Arc, time::Duration};
 
 mod test_helpers;
-use test_helpers::{now, random_topic_name, record};
-
-use crate::test_helpers::maybe_start_logging;
+use test_helpers::{maybe_start_logging, now, random_topic_name, record};
 
 #[tokio::test]
 async fn test_plain() {
@@ -449,18 +447,7 @@ async fn test_delete_records() {
     let offset_4 = offsets[0];
 
     // delete from the middle of the 2nd batch
-    match partition_client.delete_records(offset_3, 1_000).await {
-        Ok(()) => {}
-        // Redpanda does not support deletions (https://github.com/redpanda-data/redpanda/issues/1016), but we also
-        // don't wanna skip it unconditionally
-        Err(ClientError::Request(RequestError::NoVersionMatch { .. }))
-            if std::env::var("TEST_DELETE_RECORDS").is_err() =>
-        {
-            println!("Skip test_delete_records");
-            return;
-        }
-        Err(e) => panic!("Cannot delete: {e}"),
-    }
+    maybe_skip_delete!(partition_client, offset_3);
 
     // fetching data before the record fails
     let err = partition_client

tests/consumer.rs

@@ -68,6 +68,45 @@ async fn test_stream_consumer_start_at_0() {
     assert_stream_pending(&mut stream).await;
 }
 
+#[tokio::test]
+async fn test_stream_consumer_start_at_1() {
+    maybe_start_logging();
+
+    let connection = maybe_skip_kafka_integration!();
+    let client = ClientBuilder::new(vec![connection]).build().await.unwrap();
+    let controller_client = client.controller_client().await.unwrap();
+
+    let topic = random_topic_name();
+    controller_client
+        .create_topic(&topic, 1, 1, 5_000)
+        .await
+        .unwrap();
+
+    let record_1 = record(b"x");
+    let record_2 = record(b"y");
+
+    let partition_client = Arc::new(client.partition_client(&topic, 0).await.unwrap());
+    partition_client
+        .produce(
+            vec![record_1.clone(), record_2.clone()],
+            Compression::NoCompression,
+        )
+        .await
+        .unwrap();
+
+    let mut stream = StreamConsumerBuilder::new(Arc::clone(&partition_client), StartOffset::At(1))
+        .with_max_wait_ms(50)
+        .build();
+
+    // Skips first record
+    let (record_and_offset, _watermark) =
+        assert_ok(timeout(Duration::from_millis(100), stream.next()).await);
+    assert_eq!(record_and_offset.record, record_2);
+
+    // No further records
+    assert_stream_pending(&mut stream).await;
+}
+
 #[tokio::test]
 async fn test_stream_consumer_offset_out_of_range() {
     maybe_start_logging();
@@ -179,13 +218,55 @@ async fn test_stream_consumer_start_at_earliest_empty() {
 
     // Get second record
     let (record_and_offset, _) =
-        assert_ok(timeout(Duration::from_millis(100), stream.next()).await);
+        assert_ok(timeout(Duration::from_millis(200), stream.next()).await);
     assert_eq!(record_and_offset.record, record);
 
     // No further records
     assert_stream_pending(&mut stream).await;
 }
 
+#[tokio::test]
+async fn test_stream_consumer_start_at_earliest_after_deletion() {
+    maybe_start_logging();
+
+    let connection = maybe_skip_kafka_integration!();
+    let client = ClientBuilder::new(vec![connection]).build().await.unwrap();
+    let controller_client = client.controller_client().await.unwrap();
+
+    let topic = random_topic_name();
+    controller_client
+        .create_topic(&topic, 1, 1, 5_000)
+        .await
+        .unwrap();
+
+    let record_1 = record(b"x");
+    let record_2 = record(b"y");
+
+    let partition_client = Arc::new(client.partition_client(&topic, 0).await.unwrap());
+    partition_client
+        .produce(
+            vec![record_1.clone(), record_2.clone()],
+            Compression::NoCompression,
+        )
+        .await
+        .unwrap();
+
+    maybe_skip_delete!(partition_client, 1);
+
+    let mut stream =
+        StreamConsumerBuilder::new(Arc::clone(&partition_client), StartOffset::Earliest)
+            .with_max_wait_ms(50)
+            .build();
+
+    // First record skipped / deleted
+    let (record_and_offset, _) =
+        assert_ok(timeout(Duration::from_millis(100), stream.next()).await);
+    assert_eq!(record_and_offset.record, record_2);
+
+    // No further records
+    assert_stream_pending(&mut stream).await;
+}
+
 #[tokio::test]
 async fn test_stream_consumer_start_at_latest() {
     maybe_start_logging();

tests/test_helpers.rs

@@ -2,6 +2,7 @@ use parking_lot::Once;
 use rskafka::record::Record;
 use std::collections::BTreeMap;
 use time::OffsetDateTime;
+
 /// Get the testing Kafka connection string or return current scope.
 ///
 /// If `TEST_INTEGRATION` and `KAFKA_CONNECT` are set, return the Kafka connection URL to the
@@ -44,6 +45,29 @@ macro_rules! maybe_skip_kafka_integration {
     }};
 }
 
+/// Performs delete operation using.
+///
+/// This is skipped (via `return`) if the broker returns `NoVersionMatch`, except when the `TEST_DELETE_RECORDS`
+/// environment variable is set.
+///
+/// This is helpful because Redpanda does not support deletes yet, see
+/// <https://github.com/redpanda-data/redpanda/issues/1016> but we also don not want to skip these test unconditionally.
+#[macro_export]
+macro_rules! maybe_skip_delete {
+    ($partition_client:ident, $offset:expr) => {
+        match $partition_client.delete_records($offset, 1_000).await {
+            Ok(()) => {}
+            Err(rskafka::client::error::Error::Request(
+                rskafka::client::error::RequestError::NoVersionMatch { .. },
+            )) if std::env::var("TEST_DELETE_RECORDS").is_err() => {
+                println!("Skip test_delete_records");
+                return;
+            }
+            Err(e) => panic!("Cannot delete: {e}"),
+        }
+    };
+}
+
 /// Generated random topic name for testing.
 pub fn random_topic_name() -> String {
     format!("test_topic_{}", uuid::Uuid::new_v4())