dekaf: support reading documents at the requested offset

Do this by starting `Read`s up to the max size of one document before the requested offset

crates/dekaf/src/read.rs

@@ -7,7 +7,6 @@ use gazette::journal::{ReadJsonLine, ReadJsonLines};
 use gazette::{broker, journal, uuid};
 use kafka_protocol::records::{Compression, TimestampType};
 use lz4_flex::frame::BlockMode;
-use std::time::{Duration, Instant};
 
 pub struct Read {
     /// Journal offset to be served by this Read.
@@ -29,6 +28,9 @@ pub struct Read {
     // Keep these details around so we can create a new ReadRequest if we need to skip forward
     journal_name: String,
 
+    // Offset before which no documents should be emitted
+    offset_start: i64,
+
     pub(crate) rewrite_offsets_from: Option<i64>,
 }
 
@@ -47,6 +49,8 @@ pub enum ReadTarget {
     Docs(usize),
 }
 
+const OFFSET_READBACK: i64 = 2 << 25 + 1; // 64mb, single document max size
+
 impl Read {
     pub fn new(
         client: journal::Client,
@@ -61,7 +65,8 @@ impl Read {
 
         let stream = client.clone().read_json_lines(
             broker::ReadRequest {
-                offset,
+                // Start reading at least 1 document in the past
+                offset: std::cmp::max(0, offset - OFFSET_READBACK),
                 block: true,
                 journal: partition.spec.name.clone(),
                 begin_mod_time: not_before_sec as i64,
@@ -89,14 +94,15 @@ impl Read {
 
             journal_name: partition.spec.name.clone(),
             rewrite_offsets_from,
+            offset_start: offset,
         }
     }
 
     #[tracing::instrument(skip_all,fields(journal_name=self.journal_name))]
     pub async fn next_batch(
         mut self,
         target: ReadTarget,
-        timeout: Instant,
+        timeout: std::time::Instant,
     ) -> anyhow::Result<(Self, BatchResult)> {
         use kafka_protocol::records::{
             Compression, Record, RecordBatchEncoder, RecordEncodeOptions,
@@ -155,7 +161,9 @@ impl Read {
                         transient_errors = transient_errors + 1;
 
                         tracing::warn!(error = ?err, "Retrying transient read error");
-                        let delay = Duration::from_millis(rand::thread_rng().gen_range(300..2000));
+                        let delay = std::time::Duration::from_millis(
+                            rand::thread_rng().gen_range(300..2000),
+                        );
                         tokio::time::sleep(delay).await;
                         // We can retry transient errors just by continuing to poll the stream
                         continue;
@@ -182,6 +190,10 @@ impl Read {
                 ReadJsonLine::Doc { root, next_offset } => (root, next_offset),
             };
 
+            if next_offset < self.offset_start {
+                continue;
+            }
+
             let mut record_bytes: usize = 0;
 
             let Some(doc::ArchivedNode::String(uuid)) = self.uuid_ptr.query(root.get()) else {

crates/dekaf/src/session.rs

@@ -430,7 +430,7 @@ impl Session {
                             // so long as the request is still a data preview request. If not, bail out
                             Entry::Occupied(entry) => {
                                 let data_preview_state = entry.get();
-                                if fetch_offset > data_preview_state.offset
+                                if fetch_offset >= data_preview_state.offset
                                     || data_preview_state.offset - fetch_offset > 12
                                 {
                                     bail!("Session was used for fetching preview data, cannot be used for fetching non-preview data.")
@@ -1318,9 +1318,9 @@ impl Session {
             .fetch_partition_offset(partition as usize, -1)
             .await?
         {
-            // If fetch_offset is > latest_offset, this is a caught-up consumer
+            // If fetch_offset is >= latest_offset, this is a caught-up consumer
             // polling for new documents, not a data preview request.
-            if fetch_offset <= latest_offset && latest_offset - fetch_offset < 13 {
+            if fetch_offset < latest_offset && latest_offset - fetch_offset < 13 {
                 tracing::info!(
                     latest_offset,
                     diff = latest_offset - fetch_offset,