kafrust_protocol/api/
fetch.rs

1use crate::codec::{DecodeLimits, Decoder, Encoder};
2use crate::error::{Error, Result};
3use crate::header::RequestHeader;
4use crate::record_batch::{decompress_record_batch_records_with_limit, RecordBatchCompression};
5
6pub const API_KEY: i16 = 1;
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct FetchRequestV2 {
10    pub correlation_id: i32,
11    pub client_id: Option<String>,
12    pub replica_id: i32,
13    pub max_wait_ms: i32,
14    pub min_bytes: i32,
15    pub topics: Vec<FetchTopicV2>,
16}
17
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub struct FetchRequestV4 {
20    pub correlation_id: i32,
21    pub client_id: Option<String>,
22    pub replica_id: i32,
23    pub max_wait_ms: i32,
24    pub min_bytes: i32,
25    pub max_bytes: i32,
26    pub isolation_level: i8,
27    pub topics: Vec<FetchTopicV2>,
28}
29
30/// Fetch request version 11 with rack-aware read selection.
31///
32/// Version 11 keeps the non-flexible wire format used by the existing direct
33/// consumer while adding fetch-session fields and the consumer rack ID.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct FetchRequestV11 {
36    pub correlation_id: i32,
37    pub client_id: Option<String>,
38    pub replica_id: i32,
39    pub max_wait_ms: i32,
40    pub min_bytes: i32,
41    pub max_bytes: i32,
42    pub isolation_level: i8,
43    pub session_id: i32,
44    pub session_epoch: i32,
45    pub topics: Vec<FetchTopicV11>,
46    pub forgotten_topics: Vec<FetchForgottenTopicV11>,
47    pub rack_id: String,
48}
49
50/// Fetch request version 12 with flexible encoding and rack-aware read selection.
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct FetchRequestV12 {
53    pub correlation_id: i32,
54    pub client_id: Option<String>,
55    pub replica_id: i32,
56    pub max_wait_ms: i32,
57    pub min_bytes: i32,
58    pub max_bytes: i32,
59    pub isolation_level: i8,
60    pub session_id: i32,
61    pub session_epoch: i32,
62    pub topics: Vec<FetchTopicV12>,
63    pub forgotten_topics: Vec<FetchForgottenTopicV12>,
64    pub rack_id: String,
65}
66
67impl FetchRequestV12 {
68    pub fn encode(&self) -> Result<Vec<u8>> {
69        let mut encoder = Encoder::new();
70        RequestHeader {
71            api_key: API_KEY,
72            api_version: 12,
73            correlation_id: self.correlation_id,
74            client_id: self.client_id.clone(),
75        }
76        .encode_v2(&mut encoder)?;
77        encoder.write_i32(self.replica_id);
78        encoder.write_i32(self.max_wait_ms);
79        encoder.write_i32(self.min_bytes);
80        encoder.write_i32(self.max_bytes);
81        encoder.write_i8(self.isolation_level);
82        encoder.write_i32(self.session_id);
83        encoder.write_i32(self.session_epoch);
84        encoder.write_compact_array(Some(self.topics.as_slice()), |encoder, topic| {
85            topic.encode(encoder)
86        })?;
87        encoder.write_compact_array(Some(self.forgotten_topics.as_slice()), |encoder, topic| {
88            topic.encode(encoder)
89        })?;
90        encoder.write_compact_string(&self.rack_id)?;
91        encoder.write_empty_tagged_fields();
92        Ok(encoder.into_bytes())
93    }
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub struct FetchTopicV12 {
98    pub name: String,
99    pub partitions: Vec<FetchPartitionV12>,
100}
101
102impl FetchTopicV12 {
103    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
104        encoder.write_compact_string(&self.name)?;
105        encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
106            partition.encode(encoder)
107        })?;
108        encoder.write_empty_tagged_fields();
109        Ok(())
110    }
111}
112
113#[derive(Debug, Clone, PartialEq, Eq)]
114pub struct FetchPartitionV12 {
115    pub partition_index: i32,
116    pub current_leader_epoch: i32,
117    pub fetch_offset: i64,
118    pub last_fetched_epoch: i32,
119    pub log_start_offset: i64,
120    pub max_bytes: i32,
121}
122
123impl FetchPartitionV12 {
124    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
125        encoder.write_i32(self.partition_index);
126        encoder.write_i32(self.current_leader_epoch);
127        encoder.write_i64(self.fetch_offset);
128        encoder.write_i32(self.last_fetched_epoch);
129        encoder.write_i64(self.log_start_offset);
130        encoder.write_i32(self.max_bytes);
131        encoder.write_empty_tagged_fields();
132        Ok(())
133    }
134}
135
136#[derive(Debug, Clone, PartialEq, Eq)]
137pub struct FetchForgottenTopicV12 {
138    pub name: String,
139    pub partitions: Vec<i32>,
140}
141
142impl FetchForgottenTopicV12 {
143    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
144        encoder.write_compact_string(&self.name)?;
145        encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
146            encoder.write_i32(*partition);
147            Ok(())
148        })?;
149        encoder.write_empty_tagged_fields();
150        Ok(())
151    }
152}
153
154impl FetchRequestV11 {
155    pub fn encode(&self) -> Result<Vec<u8>> {
156        let mut encoder = Encoder::new();
157        RequestHeader {
158            api_key: API_KEY,
159            api_version: 11,
160            correlation_id: self.correlation_id,
161            client_id: self.client_id.clone(),
162        }
163        .encode_v1(&mut encoder)?;
164        encoder.write_i32(self.replica_id);
165        encoder.write_i32(self.max_wait_ms);
166        encoder.write_i32(self.min_bytes);
167        encoder.write_i32(self.max_bytes);
168        encoder.write_i8(self.isolation_level);
169        encoder.write_i32(self.session_id);
170        encoder.write_i32(self.session_epoch);
171        encoder.write_array(Some(self.topics.as_slice()), |encoder, topic| {
172            topic.encode(encoder)
173        })?;
174        encoder.write_array(Some(self.forgotten_topics.as_slice()), |encoder, topic| {
175            topic.encode(encoder)
176        })?;
177        encoder.write_string(&self.rack_id)?;
178        Ok(encoder.into_bytes())
179    }
180}
181
182#[derive(Debug, Clone, PartialEq, Eq)]
183pub struct FetchTopicV11 {
184    pub name: String,
185    pub partitions: Vec<FetchPartitionV11>,
186}
187
188impl FetchTopicV11 {
189    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
190        encoder.write_string(&self.name)?;
191        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
192            partition.encode(encoder)
193        })
194    }
195}
196
197#[derive(Debug, Clone, PartialEq, Eq)]
198pub struct FetchPartitionV11 {
199    pub partition_index: i32,
200    pub current_leader_epoch: i32,
201    pub fetch_offset: i64,
202    pub log_start_offset: i64,
203    pub max_bytes: i32,
204}
205
206impl FetchPartitionV11 {
207    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
208        encoder.write_i32(self.partition_index);
209        encoder.write_i32(self.current_leader_epoch);
210        encoder.write_i64(self.fetch_offset);
211        encoder.write_i64(self.log_start_offset);
212        encoder.write_i32(self.max_bytes);
213        Ok(())
214    }
215}
216
217#[derive(Debug, Clone, PartialEq, Eq)]
218pub struct FetchForgottenTopicV11 {
219    pub name: String,
220    pub partitions: Vec<i32>,
221}
222
223impl FetchForgottenTopicV11 {
224    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
225        encoder.write_string(&self.name)?;
226        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
227            encoder.write_i32(*partition);
228            Ok(())
229        })
230    }
231}
232
233impl FetchRequestV4 {
234    pub fn encode(&self) -> Result<Vec<u8>> {
235        let mut encoder = Encoder::new();
236        RequestHeader {
237            api_key: API_KEY,
238            api_version: 4,
239            correlation_id: self.correlation_id,
240            client_id: self.client_id.clone(),
241        }
242        .encode_v1(&mut encoder)?;
243        encoder.write_i32(self.replica_id);
244        encoder.write_i32(self.max_wait_ms);
245        encoder.write_i32(self.min_bytes);
246        encoder.write_i32(self.max_bytes);
247        encoder.write_i8(self.isolation_level);
248        encoder.write_array(Some(self.topics.as_slice()), |encoder, topic| {
249            topic.encode(encoder)
250        })?;
251        Ok(encoder.into_bytes())
252    }
253}
254
255impl FetchRequestV2 {
256    pub fn encode(&self) -> Result<Vec<u8>> {
257        let mut encoder = Encoder::new();
258        RequestHeader {
259            api_key: API_KEY,
260            api_version: 2,
261            correlation_id: self.correlation_id,
262            client_id: self.client_id.clone(),
263        }
264        .encode_v1(&mut encoder)?;
265        encoder.write_i32(self.replica_id);
266        encoder.write_i32(self.max_wait_ms);
267        encoder.write_i32(self.min_bytes);
268        encoder.write_array(Some(self.topics.as_slice()), |encoder, topic| {
269            topic.encode(encoder)
270        })?;
271        Ok(encoder.into_bytes())
272    }
273}
274
275#[derive(Debug, Clone, PartialEq, Eq)]
276pub struct FetchTopicV2 {
277    pub name: String,
278    pub partitions: Vec<FetchPartitionV2>,
279}
280
281impl FetchTopicV2 {
282    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
283        encoder.write_string(&self.name)?;
284        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
285            partition.encode(encoder)
286        })
287    }
288}
289
290#[derive(Debug, Clone, PartialEq, Eq)]
291pub struct FetchPartitionV2 {
292    pub partition_index: i32,
293    pub fetch_offset: i64,
294    pub max_bytes: i32,
295}
296
297impl FetchPartitionV2 {
298    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
299        encoder.write_i32(self.partition_index);
300        encoder.write_i64(self.fetch_offset);
301        encoder.write_i32(self.max_bytes);
302        Ok(())
303    }
304}
305
306#[derive(Debug, Clone, PartialEq, Eq)]
307pub struct FetchResponseV2 {
308    pub throttle_time_ms: i32,
309    pub responses: Vec<FetchTopicResponseV2>,
310}
311
312#[derive(Debug, Clone, PartialEq, Eq)]
313pub struct FetchResponseV4 {
314    pub throttle_time_ms: i32,
315    pub responses: Vec<FetchTopicResponseV4>,
316}
317
318/// Fetch response version 11 with broker-selected read replicas.
319#[derive(Debug, Clone, PartialEq, Eq)]
320pub struct FetchResponseV11 {
321    pub throttle_time_ms: i32,
322    pub error_code: i16,
323    pub session_id: i32,
324    pub responses: Vec<FetchTopicResponseV11>,
325}
326
327/// Fetch response version 12 with flexible encoding and broker-selected reads.
328#[derive(Debug, Clone, PartialEq, Eq)]
329pub struct FetchResponseV12 {
330    pub throttle_time_ms: i32,
331    pub error_code: i16,
332    pub session_id: i32,
333    pub responses: Vec<FetchTopicResponseV12>,
334}
335
336impl FetchResponseV12 {
337    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
338        Ok(Self {
339            throttle_time_ms: decoder.read_i32()?,
340            error_code: decoder.read_i16()?,
341            session_id: decoder.read_i32()?,
342            responses: decoder
343                .read_compact_array("fetch responses", FetchTopicResponseV12::decode)?
344                .unwrap_or_default(),
345        })
346        .and_then(|response| {
347            decoder.read_tagged_fields()?;
348            Ok(response)
349        })
350    }
351}
352
353#[derive(Debug, Clone, PartialEq, Eq)]
354pub struct FetchTopicResponseV12 {
355    pub name: String,
356    pub partitions: Vec<FetchPartitionResponseV12>,
357}
358
359impl FetchTopicResponseV12 {
360    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
361        let name = decoder.read_compact_string()?;
362        let partitions = decoder
363            .read_compact_array(
364                "fetch partition responses",
365                FetchPartitionResponseV12::decode,
366            )?
367            .unwrap_or_default();
368        decoder.read_tagged_fields()?;
369        Ok(Self { name, partitions })
370    }
371}
372
373#[derive(Debug, Clone, PartialEq, Eq)]
374pub struct FetchPartitionResponseV12 {
375    pub partition_index: i32,
376    pub error_code: i16,
377    pub high_watermark: i64,
378    pub last_stable_offset: i64,
379    pub log_start_offset: i64,
380    pub aborted_transactions: Vec<AbortedTransactionV12>,
381    pub preferred_read_replica: i32,
382    pub records: Vec<MessageSetRecord>,
383}
384
385impl FetchPartitionResponseV12 {
386    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
387        let limits = decoder.limits();
388        let partition_index = decoder.read_i32()?;
389        let error_code = decoder.read_i16()?;
390        let high_watermark = decoder.read_i64()?;
391        let last_stable_offset = decoder.read_i64()?;
392        let log_start_offset = decoder.read_i64()?;
393        let aborted_transactions = decoder
394            .read_compact_array("aborted transactions", AbortedTransactionV12::decode)?
395            .unwrap_or_default();
396        let preferred_read_replica = decoder.read_i32()?;
397        let records = decoder
398            .read_compact_nullable_bytes()?
399            .map(|bytes| decode_message_set(&bytes, limits))
400            .transpose()?
401            .unwrap_or_default();
402        decoder.read_tagged_fields()?;
403        Ok(Self {
404            partition_index,
405            error_code,
406            high_watermark,
407            last_stable_offset,
408            log_start_offset,
409            aborted_transactions,
410            preferred_read_replica,
411            records,
412        })
413    }
414}
415
416#[derive(Debug, Clone, PartialEq, Eq)]
417pub struct AbortedTransactionV12 {
418    pub producer_id: i64,
419    pub first_offset: i64,
420}
421
422impl AbortedTransactionV12 {
423    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
424        let producer_id = decoder.read_i64()?;
425        let first_offset = decoder.read_i64()?;
426        decoder.read_tagged_fields()?;
427        Ok(Self {
428            producer_id,
429            first_offset,
430        })
431    }
432}
433
434impl FetchResponseV11 {
435    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
436        Ok(Self {
437            throttle_time_ms: decoder.read_i32()?,
438            error_code: decoder.read_i16()?,
439            session_id: decoder.read_i32()?,
440            responses: decoder
441                .read_array("fetch responses", FetchTopicResponseV11::decode)?
442                .unwrap_or_default(),
443        })
444    }
445}
446
447#[derive(Debug, Clone, PartialEq, Eq)]
448pub struct FetchTopicResponseV11 {
449    pub name: String,
450    pub partitions: Vec<FetchPartitionResponseV11>,
451}
452
453impl FetchTopicResponseV11 {
454    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
455        Ok(Self {
456            name: decoder.read_string()?,
457            partitions: decoder
458                .read_array(
459                    "fetch partition responses",
460                    FetchPartitionResponseV11::decode,
461                )?
462                .unwrap_or_default(),
463        })
464    }
465}
466
467#[derive(Debug, Clone, PartialEq, Eq)]
468pub struct FetchPartitionResponseV11 {
469    pub partition_index: i32,
470    pub error_code: i16,
471    pub high_watermark: i64,
472    pub last_stable_offset: i64,
473    pub log_start_offset: i64,
474    pub aborted_transactions: Vec<AbortedTransactionV4>,
475    pub preferred_read_replica: i32,
476    pub records: Vec<MessageSetRecord>,
477}
478
479impl FetchPartitionResponseV11 {
480    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
481        let limits = decoder.limits();
482        Ok(Self {
483            partition_index: decoder.read_i32()?,
484            error_code: decoder.read_i16()?,
485            high_watermark: decoder.read_i64()?,
486            last_stable_offset: decoder.read_i64()?,
487            log_start_offset: decoder.read_i64()?,
488            aborted_transactions: decoder
489                .read_array("aborted transactions", AbortedTransactionV4::decode)?
490                .unwrap_or_default(),
491            preferred_read_replica: decoder.read_i32()?,
492            records: decode_message_set(&decoder.read_bytes()?, limits)?,
493        })
494    }
495}
496
497impl FetchResponseV4 {
498    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
499        Ok(Self {
500            throttle_time_ms: decoder.read_i32()?,
501            responses: decoder
502                .read_array("fetch responses", FetchTopicResponseV4::decode)?
503                .unwrap_or_default(),
504        })
505    }
506}
507
508#[derive(Debug, Clone, PartialEq, Eq)]
509pub struct FetchTopicResponseV4 {
510    pub name: String,
511    pub partitions: Vec<FetchPartitionResponseV4>,
512}
513
514impl FetchTopicResponseV4 {
515    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
516        Ok(Self {
517            name: decoder.read_string()?,
518            partitions: decoder
519                .read_array(
520                    "fetch partition responses",
521                    FetchPartitionResponseV4::decode,
522                )?
523                .unwrap_or_default(),
524        })
525    }
526}
527
528#[derive(Debug, Clone, PartialEq, Eq)]
529pub struct FetchPartitionResponseV4 {
530    pub partition_index: i32,
531    pub error_code: i16,
532    pub high_watermark: i64,
533    pub last_stable_offset: i64,
534    pub aborted_transactions: Vec<AbortedTransactionV4>,
535    pub records: Vec<MessageSetRecord>,
536}
537
538impl FetchPartitionResponseV4 {
539    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
540        let limits = decoder.limits();
541        Ok(Self {
542            partition_index: decoder.read_i32()?,
543            error_code: decoder.read_i16()?,
544            high_watermark: decoder.read_i64()?,
545            last_stable_offset: decoder.read_i64()?,
546            aborted_transactions: decoder
547                .read_array("aborted transactions", AbortedTransactionV4::decode)?
548                .unwrap_or_default(),
549            records: decode_message_set(&decoder.read_bytes()?, limits)?,
550        })
551    }
552}
553
554#[derive(Debug, Clone, PartialEq, Eq)]
555pub struct AbortedTransactionV4 {
556    pub producer_id: i64,
557    pub first_offset: i64,
558}
559
560impl AbortedTransactionV4 {
561    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
562        Ok(Self {
563            producer_id: decoder.read_i64()?,
564            first_offset: decoder.read_i64()?,
565        })
566    }
567}
568
569impl FetchResponseV2 {
570    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
571        Ok(Self {
572            throttle_time_ms: decoder.read_i32()?,
573            responses: decoder
574                .read_array("fetch responses", FetchTopicResponseV2::decode)?
575                .unwrap_or_default(),
576        })
577    }
578}
579
580#[derive(Debug, Clone, PartialEq, Eq)]
581pub struct FetchTopicResponseV2 {
582    pub name: String,
583    pub partitions: Vec<FetchPartitionResponseV2>,
584}
585
586impl FetchTopicResponseV2 {
587    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
588        Ok(Self {
589            name: decoder.read_string()?,
590            partitions: decoder
591                .read_array(
592                    "fetch partition responses",
593                    FetchPartitionResponseV2::decode,
594                )?
595                .unwrap_or_default(),
596        })
597    }
598}
599
600#[derive(Debug, Clone, PartialEq, Eq)]
601pub struct FetchPartitionResponseV2 {
602    pub partition_index: i32,
603    pub error_code: i16,
604    pub high_watermark: i64,
605    pub records: Vec<MessageSetRecord>,
606}
607
608impl FetchPartitionResponseV2 {
609    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
610        let limits = decoder.limits();
611        Ok(Self {
612            partition_index: decoder.read_i32()?,
613            error_code: decoder.read_i16()?,
614            high_watermark: decoder.read_i64()?,
615            records: decode_message_set(&decoder.read_bytes()?, limits)?,
616        })
617    }
618}
619
620#[derive(Debug, Clone, PartialEq, Eq)]
621pub struct MessageSetRecord {
622    pub offset: i64,
623    pub timestamp_ms: i64,
624    pub key: Option<Vec<u8>>,
625    pub value: Option<Vec<u8>>,
626    pub producer_id: Option<i64>,
627    pub transactional: bool,
628    pub control: bool,
629}
630
631fn decode_message_set(bytes: &[u8], limits: DecodeLimits) -> Result<Vec<MessageSetRecord>> {
632    let mut decoder = Decoder::with_limits(bytes, limits);
633    let mut records = Vec::new();
634
635    while decoder.remaining() >= 12 {
636        let offset = decoder.read_i64()?;
637        let message_size = decoder.read_i32()?;
638        if message_size < 0 {
639            return Err(Error::NegativeLength {
640                kind: "message",
641                length: message_size,
642            });
643        }
644        let message_size =
645            usize::try_from(message_size).map_err(|_| Error::LengthOverflow("message"))?;
646        // Fetch responses may end with a partial trailing message set entry.
647        if decoder.remaining() < message_size {
648            break;
649        }
650        let message = decoder.read_exact(message_size)?;
651        let decoded = decode_message_or_batch(offset, message, limits)?;
652        let total = records
653            .len()
654            .checked_add(decoded.len())
655            .ok_or(Error::LengthOverflow("fetch records"))?;
656        decoder.ensure_collection_length("fetch records", total)?;
657        records.extend(decoded);
658    }
659
660    Ok(records)
661}
662
663fn decode_message_or_batch(
664    offset: i64,
665    bytes: &[u8],
666    limits: DecodeLimits,
667) -> Result<Vec<MessageSetRecord>> {
668    match bytes.get(4).copied() {
669        Some(2) => decode_record_batch(offset, bytes, limits),
670        _ => Ok(vec![decode_message(offset, bytes, limits)?]),
671    }
672}
673
674fn decode_message(offset: i64, bytes: &[u8], limits: DecodeLimits) -> Result<MessageSetRecord> {
675    let mut decoder = Decoder::with_limits(bytes, limits);
676    let _crc = decoder.read_i32()?;
677    let magic = decoder.read_i8()?;
678    let _attributes = decoder.read_i8()?;
679    let timestamp_ms = match magic {
680        0 => -1,
681        1 => decoder.read_i64()?,
682        _ => {
683            return Err(Error::UnsupportedVersion {
684                kind: "message magic",
685                version: i16::from(magic),
686            })
687        }
688    };
689    let key = decoder.read_nullable_bytes()?;
690    let value = decoder.read_nullable_bytes()?;
691
692    Ok(MessageSetRecord {
693        offset,
694        timestamp_ms,
695        key,
696        value,
697        producer_id: None,
698        transactional: false,
699        control: false,
700    })
701}
702
703fn decode_record_batch(
704    base_offset: i64,
705    bytes: &[u8],
706    limits: DecodeLimits,
707) -> Result<Vec<MessageSetRecord>> {
708    let mut decoder = Decoder::with_limits(bytes, limits);
709    let _partition_leader_epoch = decoder.read_i32()?;
710    let magic = decoder.read_i8()?;
711    if magic != 2 {
712        return Err(Error::UnsupportedVersion {
713            kind: "record batch magic",
714            version: i16::from(magic),
715        });
716    }
717    let _crc = decoder.read_i32()?;
718    let attributes = decoder.read_i16()?;
719    let compression = RecordBatchCompression::from_attributes(attributes)?;
720    let _last_offset_delta = decoder.read_i32()?;
721    let base_timestamp = decoder.read_i64()?;
722    let _max_timestamp = decoder.read_i64()?;
723    let producer_id = decoder.read_i64()?;
724    let _producer_epoch = decoder.read_i16()?;
725    let _base_sequence = decoder.read_i32()?;
726    let record_count = decoder.read_i32()?;
727    if record_count < 0 {
728        return Err(Error::NegativeLength {
729            kind: "record batch records",
730            length: record_count,
731        });
732    }
733
734    let record_count =
735        usize::try_from(record_count).map_err(|_| Error::LengthOverflow("record batch records"))?;
736    decoder.ensure_collection_length("record batch records", record_count)?;
737    let record_bytes = if compression.is_compressed() {
738        let compressed = decoder.read_exact(decoder.remaining())?;
739        decompress_record_batch_records_with_limit(
740            compression,
741            compressed,
742            limits.max_decompressed_record_bytes(),
743        )?
744    } else {
745        if decoder.remaining() > limits.max_decompressed_record_bytes() {
746            return Err(Error::LimitExceeded {
747                kind: "decompressed record batch bytes",
748                actual: decoder.remaining(),
749                max: limits.max_decompressed_record_bytes(),
750            });
751        }
752        decoder.read_exact(decoder.remaining())?.to_vec()
753    };
754    let mut record_decoder = Decoder::with_limits(&record_bytes, limits);
755    let mut records = Vec::with_capacity(record_count);
756    for _ in 0..record_count {
757        let record_length = record_decoder.read_varint()?;
758        if record_length < 0 {
759            return Err(Error::NegativeLength {
760                kind: "record",
761                length: record_length,
762            });
763        }
764        let record_length =
765            usize::try_from(record_length).map_err(|_| Error::LengthOverflow("record"))?;
766        let record_bytes = record_decoder.read_exact(record_length)?;
767        records.push(decode_record(
768            base_offset,
769            base_timestamp,
770            producer_id,
771            attributes,
772            record_bytes,
773            limits,
774        )?);
775    }
776
777    Ok(records)
778}
779
780fn decode_record(
781    base_offset: i64,
782    base_timestamp: i64,
783    producer_id: i64,
784    batch_attributes: i16,
785    bytes: &[u8],
786    limits: DecodeLimits,
787) -> Result<MessageSetRecord> {
788    let mut decoder = Decoder::with_limits(bytes, limits);
789    let _attributes = decoder.read_i8()?;
790    let timestamp_delta = decoder.read_varlong()?;
791    let offset_delta = decoder.read_varint()?;
792    let key = decoder.read_varint_nullable_bytes()?;
793    let value = decoder.read_varint_nullable_bytes()?;
794    let header_count = decoder.read_varint()?;
795    if header_count < 0 {
796        return Err(Error::NegativeLength {
797            kind: "record headers",
798            length: header_count,
799        });
800    }
801    let header_count =
802        usize::try_from(header_count).map_err(|_| Error::LengthOverflow("record headers"))?;
803    decoder.ensure_collection_length("record headers", header_count)?;
804    for _ in 0..header_count {
805        let _header_key = decoder.read_varint_bytes()?;
806        let _header_value = decoder.read_varint_nullable_bytes()?;
807    }
808
809    Ok(MessageSetRecord {
810        offset: base_offset.saturating_add(i64::from(offset_delta)),
811        timestamp_ms: base_timestamp.saturating_add(timestamp_delta),
812        key,
813        value,
814        producer_id: (producer_id >= 0).then_some(producer_id),
815        transactional: batch_attributes & 0x10 != 0,
816        control: batch_attributes & 0x20 != 0,
817    })
818}
819
820#[cfg(test)]
821#[allow(clippy::unwrap_used)]
822mod tests {
823    use super::{
824        FetchPartitionV11, FetchPartitionV12, FetchPartitionV2, FetchRequestV11, FetchRequestV12,
825        FetchRequestV2, FetchRequestV4, FetchResponseV11, FetchResponseV12, FetchResponseV2,
826        FetchResponseV4, FetchTopicV11, FetchTopicV12, FetchTopicV2, MessageSetRecord,
827    };
828    use crate::codec::{Decoder, Encoder};
829
830    #[test]
831    fn encodes_fetch_request_v2() {
832        let request = FetchRequestV2 {
833            correlation_id: 7,
834            client_id: Some("kafrust".to_owned()),
835            replica_id: -1,
836            max_wait_ms: 500,
837            min_bytes: 1,
838            topics: vec![FetchTopicV2 {
839                name: "orders".to_owned(),
840                partitions: vec![FetchPartitionV2 {
841                    partition_index: 0,
842                    fetch_offset: 42,
843                    max_bytes: 1_048_576,
844                }],
845            }],
846        };
847
848        let bytes = request.encode().unwrap();
849        assert_eq!(&bytes[0..4], &[0, 1, 0, 2]);
850        assert!(bytes.len() > 40);
851    }
852
853    #[test]
854    fn encodes_fetch_request_v4() {
855        let request = FetchRequestV4 {
856            correlation_id: 8,
857            client_id: Some("kafrust".to_owned()),
858            replica_id: -1,
859            max_wait_ms: 500,
860            min_bytes: 1,
861            max_bytes: 1_048_576,
862            isolation_level: 0,
863            topics: vec![FetchTopicV2 {
864                name: "orders".to_owned(),
865                partitions: vec![FetchPartitionV2 {
866                    partition_index: 0,
867                    fetch_offset: 42,
868                    max_bytes: 1_048_576,
869                }],
870            }],
871        };
872
873        let bytes = request.encode().unwrap();
874        assert_eq!(&bytes[0..4], &[0, 1, 0, 4]);
875        assert_eq!(&bytes[4..8], &[0, 0, 0, 8]);
876        assert!(bytes.len() > 45);
877    }
878
879    #[test]
880    fn encodes_fetch_request_v11_with_rack_and_fetch_session_fields() {
881        let request = FetchRequestV11 {
882            correlation_id: 9,
883            client_id: Some("kafrust".to_owned()),
884            replica_id: -1,
885            max_wait_ms: 500,
886            min_bytes: 1,
887            max_bytes: 1_048_576,
888            isolation_level: 1,
889            session_id: 0,
890            session_epoch: 0,
891            topics: vec![FetchTopicV11 {
892                name: "orders".to_owned(),
893                partitions: vec![FetchPartitionV11 {
894                    partition_index: 0,
895                    current_leader_epoch: -1,
896                    fetch_offset: 42,
897                    log_start_offset: -1,
898                    max_bytes: 1_048_576,
899                }],
900            }],
901            forgotten_topics: Vec::new(),
902            rack_id: "rack-a".to_owned(),
903        };
904
905        let bytes = request.encode().unwrap();
906        let mut decoder = Decoder::new(&bytes);
907        assert_eq!(decoder.read_i16().unwrap(), 1);
908        assert_eq!(decoder.read_i16().unwrap(), 11);
909        assert_eq!(decoder.read_i32().unwrap(), 9);
910        assert_eq!(
911            decoder.read_nullable_string().unwrap().as_deref(),
912            Some("kafrust")
913        );
914        assert_eq!(decoder.read_i32().unwrap(), -1);
915        assert_eq!(decoder.read_i32().unwrap(), 500);
916        assert_eq!(decoder.read_i32().unwrap(), 1);
917        assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
918        assert_eq!(decoder.read_i8().unwrap(), 1);
919        assert_eq!(decoder.read_i32().unwrap(), 0);
920        assert_eq!(decoder.read_i32().unwrap(), 0);
921        assert_eq!(decoder.read_i32().unwrap(), 1);
922        assert_eq!(decoder.read_string().unwrap(), "orders");
923        assert_eq!(decoder.read_i32().unwrap(), 1);
924        assert_eq!(decoder.read_i32().unwrap(), 0);
925        assert_eq!(decoder.read_i32().unwrap(), -1);
926        assert_eq!(decoder.read_i64().unwrap(), 42);
927        assert_eq!(decoder.read_i64().unwrap(), -1);
928        assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
929        assert_eq!(decoder.read_i32().unwrap(), 0);
930        assert_eq!(decoder.read_string().unwrap(), "rack-a");
931        assert!(decoder.is_empty());
932    }
933
934    #[test]
935    fn decodes_fetch_response_v11_with_preferred_read_replica() {
936        let mut bytes = Encoder::new();
937        bytes.write_i32(3);
938        bytes.write_i16(0);
939        bytes.write_i32(17);
940        bytes.write_i32(1);
941        bytes.write_string("orders").unwrap();
942        bytes.write_i32(1);
943        bytes.write_i32(0);
944        bytes.write_i16(0);
945        bytes.write_i64(43);
946        bytes.write_i64(42);
947        bytes.write_i64(40);
948        bytes.write_i32(0);
949        bytes.write_i32(2);
950        bytes.write_bytes(&[]).unwrap();
951
952        let bytes = bytes.into_bytes();
953        let mut decoder = Decoder::new(&bytes);
954        let response = FetchResponseV11::decode_body(&mut decoder).unwrap();
955        let partition = &response.responses[0].partitions[0];
956
957        assert_eq!(response.throttle_time_ms, 3);
958        assert_eq!(response.session_id, 17);
959        assert_eq!(partition.log_start_offset, 40);
960        assert_eq!(partition.preferred_read_replica, 2);
961        assert!(partition.records.is_empty());
962        assert!(decoder.is_empty());
963    }
964
965    #[test]
966    fn encodes_fetch_request_v12_with_flexible_rack_fields() {
967        let request = FetchRequestV12 {
968            correlation_id: 10,
969            client_id: Some("kafrust".to_owned()),
970            replica_id: -1,
971            max_wait_ms: 500,
972            min_bytes: 1,
973            max_bytes: 1_048_576,
974            isolation_level: 1,
975            session_id: 0,
976            session_epoch: 0,
977            topics: vec![FetchTopicV12 {
978                name: "orders".to_owned(),
979                partitions: vec![FetchPartitionV12 {
980                    partition_index: 0,
981                    current_leader_epoch: -1,
982                    fetch_offset: 42,
983                    last_fetched_epoch: -1,
984                    log_start_offset: -1,
985                    max_bytes: 1_048_576,
986                }],
987            }],
988            forgotten_topics: Vec::new(),
989            rack_id: "rack-a".to_owned(),
990        };
991
992        let bytes = request.encode().unwrap();
993        let mut decoder = Decoder::new(&bytes);
994        assert_eq!(decoder.read_i16().unwrap(), 1);
995        assert_eq!(decoder.read_i16().unwrap(), 12);
996        assert_eq!(decoder.read_i32().unwrap(), 10);
997        assert_eq!(
998            decoder.read_nullable_string().unwrap().as_deref(),
999            Some("kafrust")
1000        );
1001        assert_eq!(decoder.read_unsigned_varint().unwrap(), 0);
1002        assert_eq!(decoder.read_i32().unwrap(), -1);
1003        assert_eq!(decoder.read_i32().unwrap(), 500);
1004        assert_eq!(decoder.read_i32().unwrap(), 1);
1005        assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1006        assert_eq!(decoder.read_i8().unwrap(), 1);
1007        assert_eq!(decoder.read_i32().unwrap(), 0);
1008        assert_eq!(decoder.read_i32().unwrap(), 0);
1009        assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
1010        assert_eq!(decoder.read_compact_string().unwrap(), "orders");
1011        assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
1012        assert_eq!(decoder.read_i32().unwrap(), 0);
1013        assert_eq!(decoder.read_i32().unwrap(), -1);
1014        assert_eq!(decoder.read_i64().unwrap(), 42);
1015        assert_eq!(decoder.read_i32().unwrap(), -1);
1016        assert_eq!(decoder.read_i64().unwrap(), -1);
1017        assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1018        assert_eq!(decoder.read_unsigned_varint().unwrap(), 0);
1019        assert_eq!(decoder.read_unsigned_varint().unwrap(), 0);
1020        assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
1021        assert_eq!(decoder.read_compact_string().unwrap(), "rack-a");
1022        assert_eq!(decoder.read_unsigned_varint().unwrap(), 0);
1023        assert!(decoder.is_empty());
1024    }
1025
1026    #[test]
1027    fn decodes_fetch_response_v12_with_preferred_read_replica() {
1028        let mut bytes = Encoder::new();
1029        bytes.write_i32(3);
1030        bytes.write_i16(0);
1031        bytes.write_i32(17);
1032        bytes.write_unsigned_varint(2);
1033        bytes.write_compact_string("orders").unwrap();
1034        bytes.write_unsigned_varint(2);
1035        bytes.write_i32(0);
1036        bytes.write_i16(0);
1037        bytes.write_i64(43);
1038        bytes.write_i64(42);
1039        bytes.write_i64(40);
1040        bytes.write_unsigned_varint(2);
1041        bytes.write_i64(7);
1042        bytes.write_i64(40);
1043        bytes.write_unsigned_varint(0);
1044        bytes.write_i32(2);
1045        bytes.write_compact_nullable_bytes(Some(&[])).unwrap();
1046        bytes.write_unsigned_varint(0);
1047        bytes.write_unsigned_varint(0);
1048        bytes.write_unsigned_varint(0);
1049
1050        let bytes = bytes.into_bytes();
1051        let mut decoder = Decoder::new(&bytes);
1052        let response = FetchResponseV12::decode_body(&mut decoder).unwrap();
1053        let partition = &response.responses[0].partitions[0];
1054
1055        assert_eq!(response.throttle_time_ms, 3);
1056        assert_eq!(response.session_id, 17);
1057        assert_eq!(partition.log_start_offset, 40);
1058        assert_eq!(partition.preferred_read_replica, 2);
1059        assert_eq!(partition.aborted_transactions[0].producer_id, 7);
1060        assert!(partition.records.is_empty());
1061        assert!(decoder.is_empty());
1062    }
1063
1064    #[test]
1065    fn decodes_fetch_response_v4_with_aborted_transaction() {
1066        let mut bytes = Encoder::new();
1067        bytes.write_i32(0);
1068        bytes.write_i32(1);
1069        bytes.write_string("orders").unwrap();
1070        bytes.write_i32(1);
1071        bytes.write_i32(0);
1072        bytes.write_i16(0);
1073        bytes.write_i64(43);
1074        bytes.write_i64(42);
1075        bytes.write_i32(1);
1076        bytes.write_i64(7);
1077        bytes.write_i64(40);
1078        bytes.write_bytes(&[]).unwrap();
1079        let bytes = bytes.into_bytes();
1080
1081        let mut decoder = Decoder::new(&bytes);
1082        let response = FetchResponseV4::decode_body(&mut decoder).unwrap();
1083        let partition = &response.responses[0].partitions[0];
1084
1085        assert_eq!(partition.high_watermark, 43);
1086        assert_eq!(partition.last_stable_offset, 42);
1087        assert_eq!(partition.aborted_transactions[0].producer_id, 7);
1088        assert_eq!(partition.aborted_transactions[0].first_offset, 40);
1089        assert!(partition.records.is_empty());
1090        assert!(decoder.is_empty());
1091    }
1092
1093    #[test]
1094    fn decodes_fetch_response_v2_with_message_set() {
1095        let mut message = Encoder::new();
1096        message.write_i32(0);
1097        message.write_i8(1);
1098        message.write_i8(0);
1099        message.write_i64(123);
1100        message.write_nullable_bytes(Some(b"order-1")).unwrap();
1101        message.write_nullable_bytes(Some(b"created")).unwrap();
1102        let message = message.into_bytes();
1103
1104        let mut set = Encoder::new();
1105        set.write_i64(42);
1106        set.write_i32(i32::try_from(message.len()).unwrap());
1107        set.write_raw(&message);
1108        let set = set.into_bytes();
1109
1110        let mut bytes = Encoder::new();
1111        bytes.write_i32(0);
1112        bytes.write_i32(1);
1113        bytes.write_string("orders").unwrap();
1114        bytes.write_i32(1);
1115        bytes.write_i32(0);
1116        bytes.write_i16(0);
1117        bytes.write_i64(43);
1118        bytes.write_bytes(&set).unwrap();
1119        let bytes = bytes.into_bytes();
1120
1121        let mut decoder = Decoder::new(&bytes);
1122        let response = FetchResponseV2::decode_body(&mut decoder).unwrap();
1123        let record = MessageSetRecord {
1124            offset: 42,
1125            timestamp_ms: 123,
1126            key: Some(b"order-1".to_vec()),
1127            value: Some(b"created".to_vec()),
1128            producer_id: None,
1129            transactional: false,
1130            control: false,
1131        };
1132
1133        assert_eq!(response.throttle_time_ms, 0);
1134        assert_eq!(response.responses[0].partitions[0].high_watermark, 43);
1135        assert_eq!(response.responses[0].partitions[0].records, vec![record]);
1136        assert!(decoder.is_empty());
1137    }
1138
1139    #[test]
1140    fn decodes_fetch_response_v2_ignores_partial_trailing_message_set_entry() {
1141        let mut message = Encoder::new();
1142        message.write_i32(0);
1143        message.write_i8(1);
1144        message.write_i8(0);
1145        message.write_i64(123);
1146        message.write_nullable_bytes(Some(b"order-1")).unwrap();
1147        message.write_nullable_bytes(Some(b"created")).unwrap();
1148        let message = message.into_bytes();
1149
1150        let mut set = Encoder::new();
1151        set.write_i64(42);
1152        set.write_i32(i32::try_from(message.len()).unwrap());
1153        set.write_raw(&message);
1154        set.write_i64(-1);
1155        set.write_i32(61);
1156        set.write_raw(&[0; 22]);
1157        let set = set.into_bytes();
1158
1159        let mut bytes = Encoder::new();
1160        bytes.write_i32(0);
1161        bytes.write_i32(1);
1162        bytes.write_string("orders").unwrap();
1163        bytes.write_i32(1);
1164        bytes.write_i32(0);
1165        bytes.write_i16(0);
1166        bytes.write_i64(43);
1167        bytes.write_bytes(&set).unwrap();
1168        let bytes = bytes.into_bytes();
1169
1170        let mut decoder = Decoder::new(&bytes);
1171        let response = FetchResponseV2::decode_body(&mut decoder).unwrap();
1172
1173        assert_eq!(response.responses[0].partitions[0].records.len(), 1);
1174        assert_eq!(response.responses[0].partitions[0].records[0].offset, 42);
1175        assert!(decoder.is_empty());
1176    }
1177
1178    #[test]
1179    fn decodes_fetch_response_v2_with_record_batch() {
1180        let mut record = Vec::new();
1181        record.push(0);
1182        write_varlong(&mut record, 5);
1183        write_varint(&mut record, 0);
1184        write_varint(&mut record, 7);
1185        record.extend_from_slice(b"order-1");
1186        write_varint(&mut record, 7);
1187        record.extend_from_slice(b"created");
1188        write_varint(&mut record, 0);
1189
1190        let mut batch = Encoder::new();
1191        batch.write_i32(0);
1192        batch.write_i8(2);
1193        batch.write_i32(0);
1194        batch.write_i16(0x10);
1195        batch.write_i32(0);
1196        batch.write_i64(1_000);
1197        batch.write_i64(1_005);
1198        batch.write_i64(7);
1199        batch.write_i16(-1);
1200        batch.write_i32(-1);
1201        batch.write_i32(1);
1202        let mut encoded_record = Vec::new();
1203        write_varint(&mut encoded_record, i32::try_from(record.len()).unwrap());
1204        encoded_record.extend_from_slice(&record);
1205        batch.write_raw(&encoded_record);
1206        let batch = batch.into_bytes();
1207
1208        let mut set = Encoder::new();
1209        set.write_i64(42);
1210        set.write_i32(i32::try_from(batch.len()).unwrap());
1211        set.write_raw(&batch);
1212        let set = set.into_bytes();
1213
1214        let mut bytes = Encoder::new();
1215        bytes.write_i32(0);
1216        bytes.write_i32(1);
1217        bytes.write_string("orders").unwrap();
1218        bytes.write_i32(1);
1219        bytes.write_i32(0);
1220        bytes.write_i16(0);
1221        bytes.write_i64(43);
1222        bytes.write_bytes(&set).unwrap();
1223        let bytes = bytes.into_bytes();
1224
1225        let mut decoder = Decoder::new(&bytes);
1226        let response = FetchResponseV2::decode_body(&mut decoder).unwrap();
1227        let record = MessageSetRecord {
1228            offset: 42,
1229            timestamp_ms: 1_005,
1230            key: Some(b"order-1".to_vec()),
1231            value: Some(b"created".to_vec()),
1232            producer_id: Some(7),
1233            transactional: true,
1234            control: false,
1235        };
1236
1237        assert_eq!(response.responses[0].partitions[0].records, vec![record]);
1238        assert!(decoder.is_empty());
1239    }
1240
1241    fn write_varint(output: &mut Vec<u8>, value: i32) {
1242        write_unsigned_varint(output, u64::from(((value << 1) ^ (value >> 31)) as u32));
1243    }
1244
1245    fn write_varlong(output: &mut Vec<u8>, value: i64) {
1246        write_unsigned_varint(output, ((value << 1) ^ (value >> 63)) as u64);
1247    }
1248
1249    fn write_unsigned_varint(output: &mut Vec<u8>, mut value: u64) {
1250        loop {
1251            let mut byte = (value & 0x7f) as u8;
1252            value >>= 7;
1253            if value != 0 {
1254                byte |= 0x80;
1255            }
1256            output.push(byte);
1257            if value == 0 {
1258                break;
1259            }
1260        }
1261    }
1262}
kafrust_protocol/api/fetch.rs

kafrust_protocol/api/
fetch.rs
