krabka_client_core/
fetch.rs

1//! Minimal single-partition `Fetch` helper over a raw [`Connection`].
2//!
3//! `krabka-client-consumer`'s group `Consumer` owns subscription-style
4//! consumption. This helper is the manual building block for callers that
5//! drive their own per-partition fetch loops with externally-owned offsets,
6//! for example the tiered-storage metadata-log consumer.
7
8use std::collections::{HashSet, VecDeque};
9
10use bytes::Bytes;
11use krabka_protocol::{
12    owned::{
13        fetch_request::{FetchPartition, FetchRequest, FetchTopic},
14        fetch_response::FetchResponse,
15    },
16    primitives::uuid::Uuid as WireUuid,
17    records::RecordHeader,
18};
19use krabka_units::{
20    ByteSize, Time, bytes,
21    convert::{ByteSizeExt as _, TimeExt as _},
22    mebibytes,
23};
24use refined_type::rule::GreaterI32;
25
26use crate::{connection::Connection, error::ClientError};
27
28/// Default whole-response byte limit for single-partition fetch helpers.
29pub const DEFAULT_FETCH_RESPONSE_MAX: ByteSize = mebibytes(50);
30/// Default minimum response size for a single-partition fetch.
31pub const DEFAULT_FETCH_MIN: ByteSize = bytes(1);
32/// [`DEFAULT_FETCH_MIN`] as a protocol byte count.
33const DEFAULT_FETCH_MIN_BYTES: i32 = 1;
34
35/// Positive whole-byte minimum response size for a single-partition fetch.
36#[derive(Clone, Copy, Debug, Eq, PartialEq, derive_more::Into)]
37pub struct FetchMinBytes(i32);
38
39impl FetchMinBytes {
40    /// Validate a protocol `min_bytes` value.
41    ///
42    /// # Errors
43    ///
44    /// Returns an error when `value` is zero or negative.
45    pub fn new(value: i32) -> Result<Self, String> {
46        GreaterI32::<0>::new(value)
47            .map(|value| Self(value.into_value()))
48            .map_err(|error| format!("fetch min bytes: {error}"))
49    }
50
51    /// Return the validated protocol byte count.
52    #[must_use]
53    pub const fn bytes(self) -> i32 {
54        self.0
55    }
56
57    /// Return the validated byte count as a dimensioned quantity.
58    #[must_use]
59    pub fn size(self) -> ByteSize {
60        ByteSize::from_bytes_i64(i64::from(self.0))
61    }
62}
63
64impl TryFrom<ByteSize> for FetchMinBytes {
65    type Error = String;
66
67    fn try_from(value: ByteSize) -> Result<Self, Self::Error> {
68        let bytes = value.bytes_f64();
69        if !bytes.is_finite()
70            || bytes.fract() != 0.0
71            || !(1.0..=f64::from(i32::MAX)).contains(&bytes)
72        {
73            return Err("fetch min must be a positive whole-byte value that fits i32".to_owned());
74        }
75        Self::new(value.bytes_i32())
76    }
77}
78
79impl Default for FetchMinBytes {
80    fn default() -> Self {
81        Self(DEFAULT_FETCH_MIN_BYTES)
82    }
83}
84
85/// The single live-IO dependency [`fetch_partition_with_isolation`] needs: send
86/// a typed [`FetchRequest`] and get the decoded [`FetchResponse`] back.
87///
88/// A trait hides this dependency, which mirrors `connect-postgres`'s
89/// `PgCatalog` seam. The trait keeps the request construction
90/// (`build_fetch_request`) and the response decode (`decode_fetch_response`)
91/// mockable without a socket. The trait method returns the already-decoded
92/// response, so a `mockall` mock drives every decode, offset, and error-code
93/// decision under the crate's default feature set. Only one part is
94/// un-mockable and it stays in the [`Connection`] adapter: the actual frame
95/// write and read on the wire.
96#[cfg_attr(test, mockall::automock)]
97#[async_trait::async_trait]
98pub trait FetchTransport: Send + Sync {
99    async fn fetch(&self, req: FetchRequest) -> Result<FetchResponse, ClientError>;
100}
101
102#[async_trait::async_trait]
103impl FetchTransport for Connection {
104    // cargo-mutants: thin Connection adapter; logic lives in fetch_partition_on
105    #[cfg_attr(test, mutants::skip)]
106    async fn fetch(&self, req: FetchRequest) -> Result<FetchResponse, ClientError> {
107        self.send(req).await
108    }
109}
110
111/// One record decoded from a single-partition fetch.
112#[derive(Debug, Clone, PartialEq, Eq)]
113pub struct FetchedRecord {
114    /// Absolute offset within the partition.
115    pub offset: i64,
116    /// Record key, if any.
117    pub key: Option<Bytes>,
118    /// Record value, if any.
119    pub value: Option<Bytes>,
120    /// Record timestamp (epoch millis): batch `base_timestamp` + per-record delta.
121    pub timestamp: i64,
122    /// Record headers in Kafka wire order, with duplicate keys and null values kept.
123    pub headers: Vec<FetchedHeader>,
124}
125
126/// Parameters for a single-partition fetch with an explicit isolation level.
127#[derive(Debug, Clone, Copy)]
128pub struct IsolatedFetch<'a> {
129    pub topic: &'a str,
130    pub topic_id: WireUuid,
131    pub partition: i32,
132    pub fetch_offset: i64,
133    pub max_wait: Time,
134    pub max: ByteSize,
135    pub partition_max: ByteSize,
136    pub fetch_min: FetchMinBytes,
137    pub isolation_level: i8,
138}
139
140/// Records returned by a fetch together with the next safe partition offset.
141///
142/// `next_offset` advances over every decoded batch, including control batches
143/// and batches filtered because their transaction aborted. Callers that own a
144/// fetch cursor must use `next_offset` and must not derive progress from
145/// `records`.
146#[derive(Debug, Clone, PartialEq, Eq)]
147pub struct FetchPartitionResult {
148    /// Visible records in offset order.
149    pub records: Vec<FetchedRecord>,
150    /// One past the highest decoded batch offset, if the response had v2 batches.
151    pub next_offset: Option<i64>,
152}
153
154/// One Kafka record header decoded from a fetched record.
155pub type FetchedHeader = RecordHeader;
156
157/// Fetch up to `partition_max` from `(topic, partition)` at `fetch_offset`.
158///
159/// This function decodes every v2 `RecordBatch` into [`FetchedRecord`]s and
160/// returns them in offset order. An empty result means the partition had
161/// nothing at or after `fetch_offset` within `max_wait`. This function skips
162/// legacy (non-v2) message sets.
163///
164/// # Errors
165///
166/// Returns [`ClientError`] on transport or version-negotiation failure.
167/// Returns [`ClientError::Server`] when the broker reports a non-zero
168/// partition-level `error_code`, for example `OFFSET_OUT_OF_RANGE`,
169/// `NOT_LEADER_OR_FOLLOWER`, or `UNKNOWN_TOPIC_ID`. The caller can then react
170/// instead of silently re-fetching the same offset forever.
171// cargo-mutants: thin wrapper over fetch_partition_on (tested core)
172#[cfg_attr(test, mutants::skip)]
173#[tracing::instrument(
174    level = "debug",
175    skip_all,
176    fields(topic = %topic, partition, fetch_offset),
177    err(level = "debug"),
178)]
179pub async fn fetch_partition(
180    conn: &Connection,
181    topic: &str,
182    topic_id: WireUuid,
183    partition: i32,
184    fetch_offset: i64,
185    max_wait: Time,
186    partition_max: ByteSize,
187) -> Result<Vec<FetchedRecord>, ClientError> {
188    fetch_partition_on(
189        conn,
190        topic,
191        topic_id,
192        partition,
193        fetch_offset,
194        max_wait,
195        partition_max,
196    )
197    .await
198}
199
200/// `FetchTransport`-generic body of [`fetch_partition`].
201///
202/// The public entry point is a thin `Connection` adapter, so the build and
203/// decode logic is unit-testable against a `mockall` `FetchTransport`.
204async fn fetch_partition_on<T: FetchTransport + ?Sized>(
205    conn: &T,
206    topic: &str,
207    topic_id: WireUuid,
208    partition: i32,
209    fetch_offset: i64,
210    max_wait: Time,
211    partition_max: ByteSize,
212) -> Result<Vec<FetchedRecord>, ClientError> {
213    // Default to READ_UNCOMMITTED (isolation_level = 0): every record visible.
214    fetch_partition_with_isolation_on(
215        conn,
216        IsolatedFetch {
217            topic,
218            topic_id,
219            partition,
220            fetch_offset,
221            max_wait,
222            max: DEFAULT_FETCH_RESPONSE_MAX,
223            partition_max,
224            fetch_min: FetchMinBytes::default(),
225            isolation_level: 0,
226        },
227    )
228    .await
229}
230
231/// Like [`fetch_partition`], but lets the caller set the Kafka
232/// `Fetch.isolation_level` (`0` = `READ_UNCOMMITTED`, `1` = `READ_COMMITTED`).
233///
234/// `READ_COMMITTED` restricts the result to records below the last stable
235/// offset and excludes records from aborted transactions. Exactly-once
236/// changelog restore needs this so that aborted writes are not replayed.
237///
238/// # Errors
239///
240/// Same as [`fetch_partition`].
241// cargo-mutants: thin wrapper over fetch_partition_with_isolation_on (tested core)
242#[cfg_attr(test, mutants::skip)]
243#[tracing::instrument(
244    level = "debug",
245    skip_all,
246    fields(topic = %fetch.topic, partition = fetch.partition, fetch_offset = fetch.fetch_offset, isolation_level = fetch.isolation_level),
247    err(level = "debug"),
248)]
249pub async fn fetch_partition_with_isolation(
250    conn: &Connection,
251    fetch: IsolatedFetch<'_>,
252) -> Result<Vec<FetchedRecord>, ClientError> {
253    fetch_partition_with_isolation_on(conn, fetch).await
254}
255
256/// Fetch a partition with isolation and return both visible records and cursor
257/// progress across filtered transactional and control batches.
258///
259/// In `READ_COMMITTED` mode this function applies the response's
260/// `aborted_transactions` metadata client-side, because Krabka brokers return
261/// the underlying record batches verbatim.
262///
263/// # Errors
264///
265/// Returns [`ClientError`] when the fetch request fails or the broker response
266/// is malformed.
267pub async fn fetch_partition_with_isolation_progress(
268    conn: &Connection,
269    fetch: IsolatedFetch<'_>,
270) -> Result<FetchPartitionResult, ClientError> {
271    fetch_partition_with_isolation_progress_on(conn, fetch).await
272}
273
274/// `FetchTransport`-generic body of [`fetch_partition_with_isolation`].
275///
276/// This function holds the build-request → send → decode-response logic, so
277/// it is killable against a `mockall` `FetchTransport` without a live broker
278/// socket.
279async fn fetch_partition_with_isolation_on<T: FetchTransport + ?Sized>(
280    conn: &T,
281    fetch: IsolatedFetch<'_>,
282) -> Result<Vec<FetchedRecord>, ClientError> {
283    let resp = conn.fetch(build_fetch_request(fetch)).await?;
284    Ok(fetch_partition_with_isolation_progress_response(
285        &resp,
286        fetch.partition,
287        fetch.fetch_offset,
288        fetch.isolation_level,
289    )?
290    .records)
291}
292
293async fn fetch_partition_with_isolation_progress_on<T: FetchTransport + ?Sized>(
294    conn: &T,
295    fetch: IsolatedFetch<'_>,
296) -> Result<FetchPartitionResult, ClientError> {
297    let resp = conn.fetch(build_fetch_request(fetch)).await?;
298    fetch_partition_with_isolation_progress_response(
299        &resp,
300        fetch.partition,
301        fetch.fetch_offset,
302        fetch.isolation_level,
303    )
304}
305
306fn build_fetch_request(fetch: IsolatedFetch<'_>) -> FetchRequest {
307    FetchRequest {
308        max_wait_ms: fetch.max_wait.millis_i32(),
309        min_bytes: fetch.fetch_min.bytes(),
310        max_bytes: fetch.max.bytes_i32(),
311        isolation_level: fetch.isolation_level,
312        topics: vec![FetchTopic {
313            topic: fetch.topic.to_owned(),
314            topic_id: fetch.topic_id,
315            partitions: vec![FetchPartition {
316                partition: fetch.partition,
317                fetch_offset: fetch.fetch_offset,
318                partition_max_bytes: fetch.partition_max.bytes_i32(),
319                ..Default::default()
320            }],
321            ..Default::default()
322        }],
323        ..Default::default()
324    }
325}
326
327/// Decode one partition response with client-side transaction filtering.
328fn fetch_partition_with_isolation_progress_response(
329    resp: &FetchResponse,
330    partition: i32,
331    fetch_floor: i64,
332    isolation_level: i8,
333) -> Result<FetchPartitionResult, ClientError> {
334    let mut out = Vec::new();
335    let mut next_offset = None;
336    let read_committed = isolation_level == 1;
337    for t in &resp.responses {
338        for p in &t.partitions {
339            if p.partition_index != partition {
340                continue;
341            }
342            if p.error_code != 0 {
343                return Err(ClientError::Server {
344                    error_code: p.error_code,
345                });
346            }
347            let Some(payload) = &p.records else { continue };
348            let Some(batches) = payload.as_v2() else {
349                continue;
350            };
351            let mut aborted: Vec<(i64, i64)> = if read_committed {
352                p.aborted_transactions
353                    .as_deref()
354                    .unwrap_or_default()
355                    .iter()
356                    .map(|transaction| (transaction.first_offset, transaction.producer_id))
357                    .collect()
358            } else {
359                Vec::new()
360            };
361            aborted.sort_unstable();
362            let mut aborted: VecDeque<_> = aborted.into();
363            let mut aborted_producers = HashSet::new();
364            for batch in batches {
365                next_offset = Some(
366                    next_offset
367                        .unwrap_or(fetch_floor)
368                        .max(batch.base_offset + i64::from(batch.last_offset_delta) + 1),
369                );
370                while read_committed
371                    && aborted
372                        .front()
373                        .is_some_and(|(first_offset, _)| *first_offset <= batch.base_offset)
374                {
375                    let Some((_, producer_id)) = aborted.pop_front() else {
376                        break;
377                    };
378                    aborted_producers.insert(producer_id);
379                }
380                if batch.attributes.is_control_batch() {
381                    if read_committed {
382                        aborted_producers.remove(&batch.producer_id);
383                    }
384                    continue;
385                }
386                if read_committed
387                    && batch.attributes.is_transactional()
388                    && aborted_producers.contains(&batch.producer_id)
389                {
390                    continue;
391                }
392                for r in &batch.records {
393                    let offset = batch.base_offset + i64::from(r.offset_delta);
394                    if offset < fetch_floor {
395                        continue;
396                    }
397                    out.push(FetchedRecord {
398                        offset,
399                        key: r.key.clone(),
400                        value: r.value.clone(),
401                        timestamp: batch.base_timestamp + r.timestamp_delta,
402                        headers: r.headers.clone(),
403                    });
404                }
405            }
406        }
407    }
408    out.sort_by_key(|r| r.offset);
409    Ok(FetchPartitionResult {
410        records: out,
411        next_offset,
412    })
413}
414
415#[cfg(test)]
416mod tests {
417    use assert2::assert;
418    use bytes::BufMut as _;
419    use krabka_protocol::{
420        UnknownTaggedFields,
421        owned::{
422            fetch_request::ReplicaState,
423            fetch_response::{
424                AbortedTransaction, FetchResponse, FetchableTopicResponse, PartitionData,
425            },
426        },
427        records::{Attributes, Record, RecordBatch, RecordsPayload},
428    };
429    use krabka_units::{ByteSize, bytes, kibibytes, millis};
430
431    use super::*;
432
433    fn put_signed_varint(buf: &mut Vec<u8>, value: i64) {
434        let mut encoded = ((value << 1) ^ (value >> 63)).cast_unsigned();
435        loop {
436            let byte = (encoded & 0x7f) as u8;
437            encoded >>= 7;
438            buf.push(if encoded == 0 { byte } else { byte | 0x80 });
439            if encoded == 0 {
440                break;
441            }
442        }
443    }
444
445    fn put_hand_encoded_record(buf: &mut Vec<u8>, body: &[u8]) {
446        put_signed_varint(buf, i64::try_from(body.len()).expect("record body fits"));
447        buf.extend_from_slice(body);
448    }
449
450    /// Build bytes directly from the Kafka record-batch v2 grammar. This
451    /// helper deliberately calls neither production encoder.
452    fn hand_encoded_v2_header_batch() -> RecordBatch {
453        let mut records = Vec::new();
454        let mut empty = Vec::new();
455        empty.push(0);
456        put_signed_varint(&mut empty, 0);
457        put_signed_varint(&mut empty, 0);
458        put_signed_varint(&mut empty, -1);
459        put_signed_varint(&mut empty, 5);
460        empty.extend_from_slice(b"empty");
461        put_signed_varint(&mut empty, 0);
462        put_hand_encoded_record(&mut records, &empty);
463
464        let long_key = "k".repeat(130);
465        let long_value = vec![0xa5; 130];
466        let mut headered = Vec::new();
467        headered.push(0);
468        put_signed_varint(&mut headered, 10);
469        put_signed_varint(&mut headered, 1);
470        put_signed_varint(&mut headered, -1);
471        put_signed_varint(&mut headered, 4);
472        headered.extend_from_slice(b"data");
473        put_signed_varint(&mut headered, 3);
474        put_signed_varint(&mut headered, 130);
475        headered.extend_from_slice(long_key.as_bytes());
476        put_signed_varint(&mut headered, 130);
477        headered.extend_from_slice(&long_value);
478        put_signed_varint(&mut headered, 3);
479        headered.extend_from_slice(b"dup");
480        put_signed_varint(&mut headered, -1);
481        put_signed_varint(&mut headered, 3);
482        headered.extend_from_slice(b"dup");
483        put_signed_varint(&mut headered, 0);
484        put_hand_encoded_record(&mut records, &headered);
485
486        let mut batch = bytes::BytesMut::new();
487        batch.put_i64(5);
488        batch.put_i32(0);
489        batch.put_i32(0);
490        batch.put_i8(2);
491        batch.put_u32(0);
492        batch.put_i16(0);
493        batch.put_i32(1);
494        batch.put_i64(1_000);
495        batch.put_i64(1_010);
496        batch.put_i64(-1);
497        batch.put_i16(-1);
498        batch.put_i32(-1);
499        batch.put_i32(2);
500        batch.extend_from_slice(&records);
501        let batch_len = i32::try_from(batch.len() - 12).expect("batch length fits");
502        batch[8..12].copy_from_slice(&batch_len.to_be_bytes());
503        let crc = crc32c::crc32c(&batch[21..]);
504        batch[17..21].copy_from_slice(&crc.to_be_bytes());
505        RecordBatch::decode(&mut &batch[..]).expect("hand-encoded v2 batch decodes")
506    }
507
508    fn batch_with(base_offset: i64, values: &[&[u8]]) -> RecordBatch {
509        let records = values
510            .iter()
511            .enumerate()
512            .map(|(i, v)| Record {
513                offset_delta: i32::try_from(i).unwrap(),
514                timestamp_delta: i64::try_from(i).unwrap() * 10,
515                value: Some(Bytes::copy_from_slice(v)),
516                ..Default::default()
517            })
518            .collect();
519        RecordBatch {
520            base_offset,
521            base_timestamp: 1_000,
522            last_offset_delta: i32::try_from(values.len().saturating_sub(1)).unwrap(),
523            records,
524            ..Default::default()
525        }
526    }
527
528    #[test]
529    fn decode_yields_absolute_offsets_for_the_requested_partition() {
530        // One batch starting at offset 5 on partition 0; a record on
531        // partition 1 that must be ignored when decoding partition 0.
532        let resp = FetchResponse {
533            responses: vec![FetchableTopicResponse {
534                topic: "t".into(),
535                partitions: vec![
536                    PartitionData {
537                        partition_index: 0,
538                        high_watermark: 7,
539                        records: Some(RecordsPayload::from(vec![batch_with(5, &[b"a", b"b"])])),
540                        ..Default::default()
541                    },
542                    PartitionData {
543                        partition_index: 1,
544                        high_watermark: 1,
545                        records: Some(RecordsPayload::from(vec![batch_with(0, &[b"z"])])),
546                        ..Default::default()
547                    },
548                ],
549                ..Default::default()
550            }],
551            ..Default::default()
552        };
553        let got = fetch_partition_with_isolation_progress_response(&resp, 0, 0, 0)
554            .unwrap()
555            .records;
556        assert!(
557            got == vec![
558                FetchedRecord {
559                    offset: 5,
560                    key: None,
561                    value: Some(Bytes::from_static(b"a")),
562                    timestamp: 1_000,
563                    headers: Vec::new(),
564                },
565                FetchedRecord {
566                    offset: 6,
567                    key: None,
568                    value: Some(Bytes::from_static(b"b")),
569                    timestamp: 1_010,
570                    headers: Vec::new(),
571                },
572            ]
573        );
574    }
575
576    #[test]
577    fn decode_preserves_hand_encoded_v2_header_wire_order_and_varints() {
578        let response = FetchResponse {
579            responses: vec![FetchableTopicResponse {
580                topic: "headers".into(),
581                partitions: vec![PartitionData {
582                    partition_index: 0,
583                    records: Some(RecordsPayload::from(vec![hand_encoded_v2_header_batch()])),
584                    ..Default::default()
585                }],
586                ..Default::default()
587            }],
588            ..Default::default()
589        };
590        let records = fetch_partition_with_isolation_progress_response(&response, 0, 0, 0)
591            .expect("fetch response")
592            .records;
593        assert_eq!(
594            records,
595            vec![
596                FetchedRecord {
597                    offset: 5,
598                    key: None,
599                    value: Some(Bytes::from_static(b"empty")),
600                    timestamp: 1_000,
601                    headers: Vec::new(),
602                },
603                FetchedRecord {
604                    offset: 6,
605                    key: None,
606                    value: Some(Bytes::from_static(b"data")),
607                    timestamp: 1_010,
608                    headers: vec![
609                        FetchedHeader {
610                            key: "k".repeat(130),
611                            value: Some(Bytes::from(vec![0xa5; 130])),
612                        },
613                        FetchedHeader {
614                            key: "dup".to_owned(),
615                            value: None,
616                        },
617                        FetchedHeader {
618                            key: "dup".to_owned(),
619                            value: Some(Bytes::new()),
620                        },
621                    ],
622                },
623            ]
624        );
625    }
626
627    #[test]
628    fn build_fetch_request_preserves_single_partition_settings() {
629        let topic_id = WireUuid([7; 16]);
630        let req = build_fetch_request(IsolatedFetch {
631            topic: "orders",
632            topic_id,
633            partition: 3,
634            fetch_offset: 123,
635            max_wait: millis(250),
636            max: kibibytes(96),
637            partition_max: kibibytes(64),
638            fetch_min: FetchMinBytes::try_from(bytes(17)).unwrap(),
639            isolation_level: 1,
640        });
641
642        assert!(
643            req == FetchRequest {
644                replica_id: -1,
645                max_wait_ms: 250,
646                min_bytes: 17,
647                max_bytes: 96 * 1024,
648                isolation_level: 1,
649                session_id: 0,
650                session_epoch: -1,
651                topics: vec![FetchTopic {
652                    topic: "orders".to_owned(),
653                    topic_id: WireUuid([7; 16]),
654                    partitions: vec![FetchPartition {
655                        partition: 3,
656                        current_leader_epoch: -1,
657                        fetch_offset: 123,
658                        last_fetched_epoch: -1,
659                        log_start_offset: -1,
660                        partition_max_bytes: 64 * 1024,
661                        replica_directory_id: WireUuid::ZERO,
662                        high_watermark: i64::MAX,
663                        unknown_tagged_fields: UnknownTaggedFields(vec![]),
664                    }],
665                    unknown_tagged_fields: UnknownTaggedFields(vec![]),
666                }],
667                forgotten_topics_data: vec![],
668                rack_id: String::new(),
669                cluster_id: None,
670                replica_state: ReplicaState {
671                    replica_id: -1,
672                    replica_epoch: -1,
673                    unknown_tagged_fields: UnknownTaggedFields(vec![]),
674                },
675                unknown_tagged_fields: UnknownTaggedFields(vec![]),
676            }
677        );
678    }
679
680    #[test]
681    fn fetch_min_validates_protocol_boundaries() {
682        assert!(FetchMinBytes::default().size() == bytes(1));
683        assert!(FetchMinBytes::default() == FetchMinBytes::try_from(DEFAULT_FETCH_MIN).unwrap());
684        assert!(FetchMinBytes::try_from(bytes(0)).is_err());
685        assert!(FetchMinBytes::try_from(ByteSize::from_bytes_f64(1.5)).is_err());
686        assert!(
687            FetchMinBytes::try_from(ByteSize::from_bytes_i64(i64::from(i32::MAX) + 1)).is_err()
688        );
689        assert!(FetchMinBytes::try_from(bytes(17)).unwrap().bytes() == 17);
690    }
691
692    #[test]
693    fn partition_error_code_surfaces_as_server_error() {
694        // OFFSET_OUT_OF_RANGE (1) on the requested partition must become
695        // an Err rather than an empty Vec, otherwise the caller would
696        // re-fetch the same offset forever.
697        let resp = FetchResponse {
698            responses: vec![FetchableTopicResponse {
699                topic: "t".into(),
700                partitions: vec![PartitionData {
701                    partition_index: 0,
702                    error_code: 1,
703                    high_watermark: 0,
704                    records: None,
705                    ..Default::default()
706                }],
707                ..Default::default()
708            }],
709            ..Default::default()
710        };
711        let err = fetch_partition_with_isolation_progress_response(&resp, 0, 0, 0).unwrap_err();
712        assert!(matches!(err, ClientError::Server { error_code: 1 }));
713    }
714
715    #[test]
716    fn read_committed_filters_aborted_batches_and_advances_past_them() {
717        let mut aborted_batch = batch_with(5, &[b"discard"]);
718        aborted_batch.producer_id = 9;
719        aborted_batch.attributes = Attributes(Attributes::TRANSACTIONAL_BIT);
720        let visible_batch = batch_with(6, &[b"keep"]);
721        let response = FetchResponse {
722            responses: vec![FetchableTopicResponse {
723                topic: "t".into(),
724                partitions: vec![PartitionData {
725                    partition_index: 0,
726                    aborted_transactions: Some(vec![AbortedTransaction {
727                        producer_id: 9,
728                        first_offset: 5,
729                        ..Default::default()
730                    }]),
731                    records: Some(RecordsPayload::from(vec![aborted_batch, visible_batch])),
732                    ..Default::default()
733                }],
734                ..Default::default()
735            }],
736            ..Default::default()
737        };
738
739        let result = fetch_partition_with_isolation_progress_response(&response, 0, 5, 1)
740            .expect("successful fetch response");
741
742        assert_eq!(result.records.len(), 1);
743        assert_eq!(result.records[0].offset, 6);
744        assert_eq!(result.records[0].value, Some(Bytes::from_static(b"keep")));
745        assert_eq!(result.next_offset, Some(7));
746    }
747
748    // ── socket-free end-to-end drive via MockFetchTransport ──────────────────
749    //
750    // These exercise `fetch_partition_with_isolation_on` (and the public
751    // `fetch_partition` adapter) end-to-end without a broker: the mock captures
752    // the `FetchRequest` the wrapper builds and returns a hand-built
753    // `FetchResponse`, so build-request + decode-response decisions are killable
754    // under the crate's default feature set.
755
756    /// The default `fetch_partition` path requests `isolation_level` 0 and
757    /// decodes the returned batch into absolute, offset-ordered records.
758    #[tokio::test]
759    async fn fetch_partition_on_builds_request_and_decodes_response() {
760        let topic_id = WireUuid([3; 16]);
761        let mut transport = MockFetchTransport::new();
762        transport
763            .expect_fetch()
764            .withf(move |req: &FetchRequest| {
765                // build_fetch_request wiring must be preserved on the wire.
766                req.isolation_level == 0
767                    && req.min_bytes == 1
768                    && req.max_bytes == 50 * 1024 * 1024
769                    && req.topics.len() == 1
770                    && req.topics[0].topic == "t"
771                    && req.topics[0].topic_id == topic_id
772                    && req.topics[0].partitions.len() == 1
773                    && req.topics[0].partitions[0].partition == 2
774                    && req.topics[0].partitions[0].fetch_offset == 5
775                    && req.topics[0].partitions[0].partition_max_bytes == 4096
776            })
777            .returning(|_req| {
778                Ok(FetchResponse {
779                    responses: vec![FetchableTopicResponse {
780                        topic: "t".into(),
781                        partitions: vec![PartitionData {
782                            partition_index: 2,
783                            high_watermark: 7,
784                            records: Some(RecordsPayload::from(vec![batch_with(5, &[b"a", b"b"])])),
785                            ..Default::default()
786                        }],
787                        ..Default::default()
788                    }],
789                    ..Default::default()
790                })
791            });
792
793        let got =
794            super::fetch_partition_on(&transport, "t", topic_id, 2, 5, millis(250), bytes(4096))
795                .await
796                .unwrap();
797        assert!(
798            got == vec![
799                FetchedRecord {
800                    offset: 5,
801                    key: None,
802                    value: Some(Bytes::from_static(b"a")),
803                    timestamp: 1_000,
804                    headers: Vec::new(),
805                },
806                FetchedRecord {
807                    offset: 6,
808                    key: None,
809                    value: Some(Bytes::from_static(b"b")),
810                    timestamp: 1_010,
811                    headers: Vec::new(),
812                },
813            ]
814        );
815    }
816
817    /// A caller-set isolation level (`READ_COMMITTED` = 1) must reach the wire.
818    #[tokio::test]
819    async fn fetch_partition_with_isolation_on_forwards_isolation_level() {
820        let topic_id = WireUuid([9; 16]);
821        let mut transport = MockFetchTransport::new();
822        transport
823            .expect_fetch()
824            .withf(|req: &FetchRequest| req.isolation_level == 1 && req.max_bytes == 2048)
825            .returning(|_req| Ok(FetchResponse::default()));
826
827        let got = super::fetch_partition_with_isolation_on(
828            &transport,
829            IsolatedFetch {
830                topic: "t",
831                topic_id,
832                partition: 0,
833                fetch_offset: 0,
834                max_wait: millis(100),
835                max: bytes(2048),
836                partition_max: bytes(1024),
837                fetch_min: FetchMinBytes::default(),
838                isolation_level: 1,
839            },
840        )
841        .await
842        .unwrap();
843        assert!(got.is_empty());
844    }
845
846    /// A transport error from the seam propagates unchanged.
847    #[tokio::test]
848    async fn fetch_partition_on_propagates_transport_error() {
849        let mut transport = MockFetchTransport::new();
850        transport
851            .expect_fetch()
852            .returning(|_req| Err(ClientError::Disconnected));
853
854        let err = super::fetch_partition_on(
855            &transport,
856            "t",
857            WireUuid([0; 16]),
858            0,
859            0,
860            millis(100),
861            bytes(1024),
862        )
863        .await
864        .unwrap_err();
865        assert!(matches!(err, ClientError::Disconnected));
866    }
867
868    /// A partition-level error code surfaces as `ClientError::Server` through
869    /// the full send→decode path, not only the isolated decode helper.
870    #[tokio::test]
871    async fn fetch_partition_on_surfaces_partition_error_code() {
872        let mut transport = MockFetchTransport::new();
873        transport.expect_fetch().returning(|_req| {
874            Ok(FetchResponse {
875                responses: vec![FetchableTopicResponse {
876                    topic: "t".into(),
877                    partitions: vec![PartitionData {
878                        partition_index: 0,
879                        error_code: 1,
880                        ..Default::default()
881                    }],
882                    ..Default::default()
883                }],
884                ..Default::default()
885            })
886        });
887
888        let err = super::fetch_partition_on(
889            &transport,
890            "t",
891            WireUuid([0; 16]),
892            0,
893            0,
894            millis(100),
895            bytes(1024),
896        )
897        .await
898        .unwrap_err();
899        assert!(matches!(err, ClientError::Server { error_code: 1 }));
900    }
901}
krabka_client_core/fetch.rs

krabka_client_core/
fetch.rs
