Merge pull request #150 from influxdata/crepererum/remove_redpanda_out_of_bounds_quirk

fix: remove potentially buggy Redpanda quirk

Author: kodiakhq[bot]

.circleci/config.yml

@@ -139,7 +139,7 @@ jobs:
     # setup multiple docker images (see https://circleci.com/docs/2.0/configuration-reference/#docker)
     docker:
       - image: quay.io/influxdb/rust:ci
-      - image: vectorized/redpanda:v21.11.2
+      - image: vectorized/redpanda:v22.1.4
         name: redpanda-0
         command:
           - redpanda
@@ -152,7 +152,7 @@ jobs:
           - --check=false
           - --kafka-addr redpanda-0:9092
           - --rpc-addr redpanda-0:33145
-      - image: vectorized/redpanda:v21.11.2
+      - image: vectorized/redpanda:v22.1.4
         name: redpanda-1
         command:
           - redpanda
@@ -166,7 +166,7 @@ jobs:
           - --kafka-addr redpanda-1:9092
           - --rpc-addr redpanda-1:33145
           - --seeds redpanda-0:33145
-      - image: vectorized/redpanda:v21.11.2
+      - image: vectorized/redpanda:v22.1.4
         name: redpanda-2
         command:
           - redpanda

docker-compose-redpanda.yml

@@ -1,7 +1,7 @@
 version: '3.7'
 services:
   redpanda-0:
-    image: vectorized/redpanda:v21.11.2
+    image: vectorized/redpanda:v22.1.4
     container_name: redpanda-0
     ports:
       - '9010:9010'
@@ -19,7 +19,7 @@ services:
       - --rpc-addr 0.0.0.0:33145
       - --advertise-rpc-addr redpanda-0:33145
   redpanda-1:
-    image: vectorized/redpanda:v21.11.2
+    image: vectorized/redpanda:v22.1.4
     container_name: redpanda-1
     ports:
       - '9011:9011'
@@ -38,7 +38,7 @@ services:
       - --rpc-addr 0.0.0.0:33146
       - --advertise-rpc-addr redpanda-1:33146
   redpanda-2:
-    image: vectorized/redpanda:v21.11.2
+    image: vectorized/redpanda:v22.1.4
     container_name: redpanda-2
     ports:
       - '9012:9012'

src/client/partition.rs

@@ -24,7 +24,7 @@ use std::ops::{ControlFlow, Deref, Range};
 use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio::sync::Mutex;
-use tracing::{error, info, warn};
+use tracing::{error, info};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum Compression {
@@ -146,18 +146,6 @@ 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, offset)?;
 
         Ok((records, partition.high_watermark.0))

tests/client.rs

@@ -125,17 +125,35 @@ async fn test_socks5() {
     let connection = maybe_skip_kafka_integration!();
     // e.g. "localhost:1080"
     let proxy = maybe_skip_SOCKS_PROXY!();
+    let topic_name = random_topic_name();
 
     let client = ClientBuilder::new(connection)
         .socks5_proxy(proxy)
         .build()
         .await
         .unwrap();
-    let partition_client = client.partition_client("myorg_mybucket", 0).unwrap();
+
+    let controller_client = client.controller_client().unwrap();
+    controller_client
+        .create_topic(&topic_name, 1, 1, 5_000)
+        .await
+        .unwrap();
+
+    let partition_client = client.partition_client(topic_name, 0).unwrap();
+
+    let record = record(b"");
     partition_client
+        .produce(vec![record.clone()], Compression::NoCompression)
+        .await
+        .unwrap();
+
+    let (mut records, _watermark) = partition_client
         .fetch_records(0, 1..10_000_001, 1_000)
         .await
         .unwrap();
+    assert_eq!(records.len(), 1);
+    let record2 = records.remove(0).record;
+    assert_eq!(record, record2);
 }
 
 #[tokio::test]