fix: heuristic to detect OffsetOutOfRange for redpanda

Author: crepererum

src/client/partition.rs

@@ -23,7 +23,7 @@ use std::ops::{ControlFlow, Deref, Range};
 use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio::sync::Mutex;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum Compression {
@@ -124,6 +124,18 @@ impl PartitionClient {
         })
         .await?;
 
+        // Redpanda never sends OffsetOutOfRange even when it should. "Luckily" it does not support deletions so we can
+        // implement a simple heuristic.
+        if partition.high_watermark.0 < offset {
+            warn!(
+                "This message looks like Redpanda wants to report a OffsetOutOfRange but doesn't."
+            );
+            return Err(Error::ServerError(
+                ProtocolError::OffsetOutOfRange,
+                String::from("Offset out of range"),
+            ));
+        }
+
         let records = extract_records(partition.records.0)?;
 
         Ok((records, partition.high_watermark.0))

tests/client.rs

@@ -1,3 +1,4 @@
+use assert_matches::assert_matches;
 use rskafka::{
     client::{
         error::{Error as ClientError, ProtocolError},
@@ -157,6 +158,63 @@ async fn test_produce_empty() {
         .unwrap();
 }
 
+#[tokio::test]
+async fn test_consume_empty() {
+    maybe_start_logging();
+
+    let connection = maybe_skip_kafka_integration!();
+    let topic_name = random_topic_name();
+    let n_partitions = 2;
+
+    let client = ClientBuilder::new(vec![connection]).build().await.unwrap();
+    let controller_client = client.controller_client().await.unwrap();
+    controller_client
+        .create_topic(&topic_name, n_partitions, 1, 5_000)
+        .await
+        .unwrap();
+
+    let partition_client = client.partition_client(&topic_name, 1).await.unwrap();
+    let (records, watermark) = partition_client
+        .fetch_records(0, 1..10_000, 1_000)
+        .await
+        .unwrap();
+    assert!(records.is_empty());
+    assert_eq!(watermark, 0);
+}
+
+#[tokio::test]
+async fn test_consume_offset_out_of_range() {
+    maybe_start_logging();
+
+    let connection = maybe_skip_kafka_integration!();
+    let topic_name = random_topic_name();
+    let n_partitions = 2;
+
+    let client = ClientBuilder::new(vec![connection]).build().await.unwrap();
+    let controller_client = client.controller_client().await.unwrap();
+    controller_client
+        .create_topic(&topic_name, n_partitions, 1, 5_000)
+        .await
+        .unwrap();
+
+    let partition_client = client.partition_client(&topic_name, 1).await.unwrap();
+    let record = record();
+    let offsets = partition_client
+        .produce(vec![record], Compression::NoCompression)
+        .await
+        .unwrap();
+    let offset = offsets[0];
+
+    let err = partition_client
+        .fetch_records(offset + 2, 1..10_000, 1_000)
+        .await
+        .unwrap_err();
+    assert_matches!(
+        err,
+        ClientError::ServerError(ProtocolError::OffsetOutOfRange, _)
+    );
+}
+
 #[tokio::test]
 async fn test_get_high_watermark() {
     maybe_start_logging();

tests/consumer.rs

@@ -37,7 +37,9 @@ async fn test_stream_consumer() {
         .await
         .unwrap();
 
-    let mut stream = StreamConsumerBuilder::new(Arc::clone(&partition_client), 0).build();
+    let mut stream = StreamConsumerBuilder::new(Arc::clone(&partition_client), 0)
+        .with_max_wait_ms(50)
+        .build();
 
     let assert_ok =
         |r: Result<Option<<StreamConsumer as Stream>::Item>, tokio::time::error::Elapsed>| {
@@ -50,7 +52,7 @@ async fn test_stream_consumer() {
     assert_ok(timeout(Duration::from_millis(100), stream.next()).await);
 
     // No further records
-    timeout(Duration::from_millis(100), stream.next())
+    timeout(Duration::from_millis(200), stream.next())
         .await
         .expect_err("timeout");