Merge pull request #163 from influxdata/crepererum/offset_in_error

feat: add request offset to failed fetch requests

Author: kodiakhq[bot]

src/client/error.rs

@@ -12,6 +12,19 @@ pub enum RequestContext {
 
     /// Error is specific to a partition (indexed via topic name and partition ID).
     Partition(String, i32),
+
+    /// Error is specific to a fetch request.
+    #[non_exhaustive]
+    Fetch {
+        /// Topic name.
+        topic_name: String,
+
+        /// Partition ID.
+        partition_id: i32,
+
+        /// Offset used during the request.
+        offset: i64,
+    },
 }
 
 /// Usable broker data for [`Error::ServerError`].

src/client/partition.rs

@@ -154,7 +154,7 @@ impl PartitionClient {
 
         let partition = maybe_retry(&self.backoff_config, self, "fetch_records", || async move {
             let response = self.get().await?.request(&request).await?;
-            process_fetch_response(self.partition, &self.topic, response)
+            process_fetch_response(self.partition, &self.topic, response, offset)
         })
         .await?;
 
@@ -546,6 +546,7 @@ fn process_fetch_response(
     partition: i32,
     topic: &str,
     response: FetchResponse,
+    request_offset: i64,
 ) -> Result<FetchResponsePartition> {
     let response_topic = response
         .responses
@@ -575,7 +576,11 @@ fn process_fetch_response(
         return Err(Error::ServerError {
             protocol_error: err,
             error_message: None,
-            request: RequestContext::Partition(topic.to_owned(), partition),
+            request: RequestContext::Fetch {
+                topic_name: topic.to_owned(),
+                partition_id: partition,
+                offset: request_offset,
+            },
             response: Some(ServerErrorResponse::PartitionFetchState {
                 high_watermark: response_partition.high_watermark.0,
                 last_stable_offset: response_partition.last_stable_offset.map(|x| x.0),