1use crate::api::produce::RecordBatchHeader;
2use crate::codec::{DecodeLimits, Decoder, Encoder};
3use crate::error::{Error, Result};
4use crate::header::RequestHeader;
5use crate::record_batch::{decompress_record_batch_records_with_limit, RecordBatchCompression};
6
7pub const API_KEY: i16 = 1;
8
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct FetchRequestV2 {
11 pub correlation_id: i32,
12 pub client_id: Option<String>,
13 pub replica_id: i32,
14 pub max_wait_ms: i32,
15 pub min_bytes: i32,
16 pub topics: Vec<FetchTopicV2>,
17}
18
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub struct FetchRequestV4 {
21 pub correlation_id: i32,
22 pub client_id: Option<String>,
23 pub replica_id: i32,
24 pub max_wait_ms: i32,
25 pub min_bytes: i32,
26 pub max_bytes: i32,
27 pub isolation_level: i8,
28 pub topics: Vec<FetchTopicV2>,
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct FetchRequestV11 {
37 pub correlation_id: i32,
38 pub client_id: Option<String>,
39 pub replica_id: i32,
40 pub max_wait_ms: i32,
41 pub min_bytes: i32,
42 pub max_bytes: i32,
43 pub isolation_level: i8,
44 pub session_id: i32,
45 pub session_epoch: i32,
46 pub topics: Vec<FetchTopicV11>,
47 pub forgotten_topics: Vec<FetchForgottenTopicV11>,
48 pub rack_id: String,
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct FetchRequestV12 {
54 pub correlation_id: i32,
55 pub client_id: Option<String>,
56 pub replica_id: i32,
57 pub max_wait_ms: i32,
58 pub min_bytes: i32,
59 pub max_bytes: i32,
60 pub isolation_level: i8,
61 pub session_id: i32,
62 pub session_epoch: i32,
63 pub topics: Vec<FetchTopicV12>,
64 pub forgotten_topics: Vec<FetchForgottenTopicV12>,
65 pub rack_id: String,
66}
67
68#[derive(Debug, Clone, PartialEq, Eq)]
75pub struct FetchRequestV13 {
76 pub correlation_id: i32,
77 pub client_id: Option<String>,
78 pub cluster_id: Option<String>,
79 pub replica_id: i32,
80 pub max_wait_ms: i32,
81 pub min_bytes: i32,
82 pub max_bytes: i32,
83 pub isolation_level: i8,
84 pub session_id: i32,
85 pub session_epoch: i32,
86 pub topics: Vec<FetchTopicV13>,
87 pub forgotten_topics: Vec<FetchForgottenTopicV13>,
88 pub rack_id: String,
89}
90
91impl FetchRequestV12 {
92 pub fn encode(&self) -> Result<Vec<u8>> {
93 let mut encoder = Encoder::new();
94 RequestHeader {
95 api_key: API_KEY,
96 api_version: 12,
97 correlation_id: self.correlation_id,
98 client_id: self.client_id.clone(),
99 }
100 .encode_v2(&mut encoder)?;
101 encoder.write_i32(self.replica_id);
102 encoder.write_i32(self.max_wait_ms);
103 encoder.write_i32(self.min_bytes);
104 encoder.write_i32(self.max_bytes);
105 encoder.write_i8(self.isolation_level);
106 encoder.write_i32(self.session_id);
107 encoder.write_i32(self.session_epoch);
108 encoder.write_compact_array(Some(self.topics.as_slice()), |encoder, topic| {
109 topic.encode(encoder)
110 })?;
111 encoder.write_compact_array(Some(self.forgotten_topics.as_slice()), |encoder, topic| {
112 topic.encode(encoder)
113 })?;
114 encoder.write_compact_string(&self.rack_id)?;
115 encoder.write_empty_tagged_fields();
116 Ok(encoder.into_bytes())
117 }
118}
119
120impl FetchRequestV13 {
121 pub fn encode(&self) -> Result<Vec<u8>> {
122 self.encode_version(13)
123 }
124
125 fn encode_version(&self, api_version: i16) -> Result<Vec<u8>> {
126 let mut encoder = Encoder::new();
127 RequestHeader {
128 api_key: API_KEY,
129 api_version,
130 correlation_id: self.correlation_id,
131 client_id: self.client_id.clone(),
132 }
133 .encode_v2(&mut encoder)?;
134 encoder.write_i32(self.replica_id);
135 encoder.write_i32(self.max_wait_ms);
136 encoder.write_i32(self.min_bytes);
137 encoder.write_i32(self.max_bytes);
138 encoder.write_i8(self.isolation_level);
139 encoder.write_i32(self.session_id);
140 encoder.write_i32(self.session_epoch);
141 encoder.write_compact_array(Some(self.topics.as_slice()), |encoder, topic| {
142 topic.encode(encoder)
143 })?;
144 encoder.write_compact_array(Some(self.forgotten_topics.as_slice()), |encoder, topic| {
145 topic.encode(encoder)
146 })?;
147 encoder.write_compact_string(&self.rack_id)?;
148 write_cluster_id_tag(&mut encoder, self.cluster_id.as_deref())?;
149 Ok(encoder.into_bytes())
150 }
151}
152
153#[derive(Debug, Clone, PartialEq, Eq)]
156pub struct FetchRequestV14 {
157 pub correlation_id: i32,
158 pub client_id: Option<String>,
159 pub cluster_id: Option<String>,
160 pub replica_id: i32,
161 pub max_wait_ms: i32,
162 pub min_bytes: i32,
163 pub max_bytes: i32,
164 pub isolation_level: i8,
165 pub session_id: i32,
166 pub session_epoch: i32,
167 pub topics: Vec<FetchTopicV14>,
168 pub forgotten_topics: Vec<FetchForgottenTopicV14>,
169 pub rack_id: String,
170}
171
172impl FetchRequestV14 {
173 pub fn encode(&self) -> Result<Vec<u8>> {
174 FetchRequestV13 {
175 correlation_id: self.correlation_id,
176 client_id: self.client_id.clone(),
177 cluster_id: self.cluster_id.clone(),
178 replica_id: self.replica_id,
179 max_wait_ms: self.max_wait_ms,
180 min_bytes: self.min_bytes,
181 max_bytes: self.max_bytes,
182 isolation_level: self.isolation_level,
183 session_id: self.session_id,
184 session_epoch: self.session_epoch,
185 topics: self.topics.clone(),
186 forgotten_topics: self.forgotten_topics.clone(),
187 rack_id: self.rack_id.clone(),
188 }
189 .encode_version(14)
190 }
191}
192
193#[derive(Debug, Clone, PartialEq, Eq)]
195pub struct FetchReplicaStateV15 {
196 pub replica_id: i32,
197 pub replica_epoch: i64,
198}
199
200#[derive(Debug, Clone, PartialEq, Eq)]
202pub struct FetchRequestV15 {
203 pub correlation_id: i32,
204 pub client_id: Option<String>,
205 pub cluster_id: Option<String>,
206 pub replica_state: Option<FetchReplicaStateV15>,
207 pub max_wait_ms: i32,
208 pub min_bytes: i32,
209 pub max_bytes: i32,
210 pub isolation_level: i8,
211 pub session_id: i32,
212 pub session_epoch: i32,
213 pub topics: Vec<FetchTopicV15>,
214 pub forgotten_topics: Vec<FetchForgottenTopicV15>,
215 pub rack_id: String,
216}
217
218impl FetchRequestV15 {
219 pub fn encode(&self) -> Result<Vec<u8>> {
220 self.encode_version(15)
221 }
222
223 fn encode_version(&self, api_version: i16) -> Result<Vec<u8>> {
224 let mut encoder = Encoder::new();
225 RequestHeader {
226 api_key: API_KEY,
227 api_version,
228 correlation_id: self.correlation_id,
229 client_id: self.client_id.clone(),
230 }
231 .encode_v2(&mut encoder)?;
232 encoder.write_i32(self.max_wait_ms);
233 encoder.write_i32(self.min_bytes);
234 encoder.write_i32(self.max_bytes);
235 encoder.write_i8(self.isolation_level);
236 encoder.write_i32(self.session_id);
237 encoder.write_i32(self.session_epoch);
238 encoder.write_compact_array(Some(self.topics.as_slice()), |encoder, topic| {
239 topic.encode(encoder)
240 })?;
241 encoder.write_compact_array(Some(self.forgotten_topics.as_slice()), |encoder, topic| {
242 topic.encode(encoder)
243 })?;
244 encoder.write_compact_string(&self.rack_id)?;
245 write_fetch_request_tags(
246 &mut encoder,
247 self.cluster_id.as_deref(),
248 self.replica_state.as_ref(),
249 )?;
250 Ok(encoder.into_bytes())
251 }
252}
253
254#[derive(Debug, Clone, PartialEq, Eq)]
256pub struct FetchRequestV16 {
257 pub correlation_id: i32,
258 pub client_id: Option<String>,
259 pub cluster_id: Option<String>,
260 pub replica_state: Option<FetchReplicaStateV15>,
261 pub max_wait_ms: i32,
262 pub min_bytes: i32,
263 pub max_bytes: i32,
264 pub isolation_level: i8,
265 pub session_id: i32,
266 pub session_epoch: i32,
267 pub topics: Vec<FetchTopicV16>,
268 pub forgotten_topics: Vec<FetchForgottenTopicV16>,
269 pub rack_id: String,
270}
271
272impl FetchRequestV16 {
273 pub fn encode(&self) -> Result<Vec<u8>> {
274 FetchRequestV15 {
275 correlation_id: self.correlation_id,
276 client_id: self.client_id.clone(),
277 cluster_id: self.cluster_id.clone(),
278 replica_state: self.replica_state.clone(),
279 max_wait_ms: self.max_wait_ms,
280 min_bytes: self.min_bytes,
281 max_bytes: self.max_bytes,
282 isolation_level: self.isolation_level,
283 session_id: self.session_id,
284 session_epoch: self.session_epoch,
285 topics: self.topics.clone(),
286 forgotten_topics: self.forgotten_topics.clone(),
287 rack_id: self.rack_id.clone(),
288 }
289 .encode_version(16)
290 }
291}
292
293#[derive(Debug, Clone, PartialEq, Eq)]
295pub struct FetchRequestV17 {
296 pub correlation_id: i32,
297 pub client_id: Option<String>,
298 pub cluster_id: Option<String>,
299 pub replica_state: Option<FetchReplicaStateV15>,
300 pub max_wait_ms: i32,
301 pub min_bytes: i32,
302 pub max_bytes: i32,
303 pub isolation_level: i8,
304 pub session_id: i32,
305 pub session_epoch: i32,
306 pub topics: Vec<FetchTopicV17>,
307 pub forgotten_topics: Vec<FetchForgottenTopicV17>,
308 pub rack_id: String,
309}
310
311impl FetchRequestV17 {
312 pub fn encode(&self) -> Result<Vec<u8>> {
313 encode_fetch_v17_plus_request(
314 17,
315 self.correlation_id,
316 self.client_id.clone(),
317 self.cluster_id.as_deref(),
318 self.replica_state.as_ref(),
319 self.max_wait_ms,
320 self.min_bytes,
321 self.max_bytes,
322 self.isolation_level,
323 self.session_id,
324 self.session_epoch,
325 &self.topics,
326 |encoder, topic| topic.encode(encoder),
327 &self.forgotten_topics,
328 self.rack_id.as_str(),
329 )
330 }
331}
332
333#[derive(Debug, Clone, PartialEq, Eq)]
335pub struct FetchRequestV18 {
336 pub correlation_id: i32,
337 pub client_id: Option<String>,
338 pub cluster_id: Option<String>,
339 pub replica_state: Option<FetchReplicaStateV15>,
340 pub max_wait_ms: i32,
341 pub min_bytes: i32,
342 pub max_bytes: i32,
343 pub isolation_level: i8,
344 pub session_id: i32,
345 pub session_epoch: i32,
346 pub topics: Vec<FetchTopicV18>,
347 pub forgotten_topics: Vec<FetchForgottenTopicV18>,
348 pub rack_id: String,
349}
350
351impl FetchRequestV18 {
352 pub fn encode(&self) -> Result<Vec<u8>> {
353 encode_fetch_v17_plus_request(
354 18,
355 self.correlation_id,
356 self.client_id.clone(),
357 self.cluster_id.as_deref(),
358 self.replica_state.as_ref(),
359 self.max_wait_ms,
360 self.min_bytes,
361 self.max_bytes,
362 self.isolation_level,
363 self.session_id,
364 self.session_epoch,
365 &self.topics,
366 |encoder, topic| topic.encode(encoder),
367 &self.forgotten_topics,
368 self.rack_id.as_str(),
369 )
370 }
371}
372
373#[allow(clippy::too_many_arguments)]
375fn encode_fetch_v17_plus_request<T, F>(
376 api_version: i16,
377 correlation_id: i32,
378 client_id: Option<String>,
379 cluster_id: Option<&str>,
380 replica_state: Option<&FetchReplicaStateV15>,
381 max_wait_ms: i32,
382 min_bytes: i32,
383 max_bytes: i32,
384 isolation_level: i8,
385 session_id: i32,
386 session_epoch: i32,
387 topics: &[T],
388 mut encode_topic: F,
389 forgotten_topics: &[FetchForgottenTopicV13],
390 rack_id: &str,
391) -> Result<Vec<u8>>
392where
393 F: FnMut(&mut Encoder, &T) -> Result<()>,
394{
395 let mut encoder = Encoder::new();
396 RequestHeader {
397 api_key: API_KEY,
398 api_version,
399 correlation_id,
400 client_id,
401 }
402 .encode_v2(&mut encoder)?;
403 encoder.write_i32(max_wait_ms);
404 encoder.write_i32(min_bytes);
405 encoder.write_i32(max_bytes);
406 encoder.write_i8(isolation_level);
407 encoder.write_i32(session_id);
408 encoder.write_i32(session_epoch);
409 encoder.write_compact_array(Some(topics), |encoder, topic| encode_topic(encoder, topic))?;
410 encoder.write_compact_array(Some(forgotten_topics), |encoder, topic| {
411 topic.encode(encoder)
412 })?;
413 encoder.write_compact_string(rack_id)?;
414 write_fetch_request_tags(&mut encoder, cluster_id, replica_state)?;
415 Ok(encoder.into_bytes())
416}
417
418fn write_cluster_id_tag(encoder: &mut Encoder, cluster_id: Option<&str>) -> Result<()> {
419 write_fetch_request_tags(encoder, cluster_id, None)
420}
421
422fn write_fetch_request_tags(
423 encoder: &mut Encoder,
424 cluster_id: Option<&str>,
425 replica_state: Option<&FetchReplicaStateV15>,
426) -> Result<()> {
427 let mut fields = Vec::new();
428 if let Some(cluster_id) = cluster_id {
429 let mut value = Encoder::new();
430 value.write_compact_string(cluster_id)?;
431 fields.push((0, value.into_bytes()));
432 }
433 if let Some(replica_state) = replica_state {
434 let mut value = Encoder::new();
435 value.write_i32(replica_state.replica_id);
436 value.write_i64(replica_state.replica_epoch);
437 fields.push((1, value.into_bytes()));
438 }
439 let field_count =
440 u32::try_from(fields.len()).map_err(|_| Error::LengthOverflow("tagged fields"))?;
441 encoder.write_unsigned_varint(field_count);
442 for (tag, value) in fields {
443 encoder.write_unsigned_varint(tag);
444 let value_len =
445 u32::try_from(value.len()).map_err(|_| Error::LengthOverflow("tagged field"))?;
446 encoder.write_unsigned_varint(value_len);
447 encoder.write_raw(&value);
448 }
449 Ok(())
450}
451
452fn write_fetch_partition_v17_tags(
453 encoder: &mut Encoder,
454 replica_directory_id: Option<&[u8; 16]>,
455 high_watermark: Option<i64>,
456) {
457 let tag_count = u32::from(replica_directory_id.is_some()) + u32::from(high_watermark.is_some());
458 encoder.write_unsigned_varint(tag_count);
459 if let Some(replica_directory_id) = replica_directory_id {
460 encoder.write_unsigned_varint(0);
461 encoder.write_unsigned_varint(16);
462 encoder.write_uuid(replica_directory_id);
463 }
464 if let Some(high_watermark) = high_watermark {
465 encoder.write_unsigned_varint(1);
466 encoder.write_unsigned_varint(8);
467 encoder.write_i64(high_watermark);
468 }
469}
470
471#[derive(Debug, Clone, PartialEq, Eq)]
472pub struct FetchTopicV12 {
473 pub name: String,
474 pub partitions: Vec<FetchPartitionV12>,
475}
476
477impl FetchTopicV12 {
478 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
479 encoder.write_compact_string(&self.name)?;
480 encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
481 partition.encode(encoder)
482 })?;
483 encoder.write_empty_tagged_fields();
484 Ok(())
485 }
486}
487
488#[derive(Debug, Clone, PartialEq, Eq)]
489pub struct FetchPartitionV12 {
490 pub partition_index: i32,
491 pub current_leader_epoch: i32,
492 pub fetch_offset: i64,
493 pub last_fetched_epoch: i32,
494 pub log_start_offset: i64,
495 pub max_bytes: i32,
496}
497
498impl FetchPartitionV12 {
499 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
500 encoder.write_i32(self.partition_index);
501 encoder.write_i32(self.current_leader_epoch);
502 encoder.write_i64(self.fetch_offset);
503 encoder.write_i32(self.last_fetched_epoch);
504 encoder.write_i64(self.log_start_offset);
505 encoder.write_i32(self.max_bytes);
506 encoder.write_empty_tagged_fields();
507 Ok(())
508 }
509}
510
511#[derive(Debug, Clone, PartialEq, Eq)]
512pub struct FetchForgottenTopicV12 {
513 pub name: String,
514 pub partitions: Vec<i32>,
515}
516
517#[derive(Debug, Clone, PartialEq, Eq)]
518pub struct FetchTopicV13 {
519 pub topic_id: [u8; 16],
520 pub partitions: Vec<FetchPartitionV13>,
521}
522
523impl FetchTopicV13 {
524 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
525 encoder.write_uuid(&self.topic_id);
526 encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
527 partition.encode(encoder)
528 })?;
529 encoder.write_empty_tagged_fields();
530 Ok(())
531 }
532}
533
534pub type FetchPartitionV13 = FetchPartitionV12;
535pub type FetchTopicV14 = FetchTopicV13;
536pub type FetchPartitionV14 = FetchPartitionV13;
537pub type FetchTopicV16 = FetchTopicV13;
538pub type FetchPartitionV16 = FetchPartitionV13;
539pub type FetchTopicV15 = FetchTopicV13;
540pub type FetchPartitionV15 = FetchPartitionV13;
541
542#[derive(Debug, Clone, PartialEq, Eq)]
543pub struct FetchTopicV17 {
544 pub topic_id: [u8; 16],
545 pub partitions: Vec<FetchPartitionV17>,
546}
547
548impl FetchTopicV17 {
549 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
550 encoder.write_uuid(&self.topic_id);
551 encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
552 partition.encode(encoder)
553 })?;
554 encoder.write_empty_tagged_fields();
555 Ok(())
556 }
557}
558
559#[derive(Debug, Clone, PartialEq, Eq)]
560pub struct FetchPartitionV17 {
561 pub partition_index: i32,
562 pub current_leader_epoch: i32,
563 pub fetch_offset: i64,
564 pub last_fetched_epoch: i32,
565 pub log_start_offset: i64,
566 pub max_bytes: i32,
567 pub replica_directory_id: Option<[u8; 16]>,
568}
569
570impl FetchPartitionV17 {
571 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
572 encoder.write_i32(self.partition_index);
573 encoder.write_i32(self.current_leader_epoch);
574 encoder.write_i64(self.fetch_offset);
575 encoder.write_i32(self.last_fetched_epoch);
576 encoder.write_i64(self.log_start_offset);
577 encoder.write_i32(self.max_bytes);
578 write_fetch_partition_v17_tags(encoder, self.replica_directory_id.as_ref(), None);
579 Ok(())
580 }
581}
582
583pub type FetchForgottenTopicV16 = FetchForgottenTopicV13;
584pub type FetchForgottenTopicV17 = FetchForgottenTopicV13;
585
586#[derive(Debug, Clone, PartialEq, Eq)]
587pub struct FetchTopicV18 {
588 pub topic_id: [u8; 16],
589 pub partitions: Vec<FetchPartitionV18>,
590}
591
592impl FetchTopicV18 {
593 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
594 encoder.write_uuid(&self.topic_id);
595 encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
596 partition.encode(encoder)
597 })?;
598 encoder.write_empty_tagged_fields();
599 Ok(())
600 }
601}
602
603#[derive(Debug, Clone, PartialEq, Eq)]
604pub struct FetchPartitionV18 {
605 pub partition_index: i32,
606 pub current_leader_epoch: i32,
607 pub fetch_offset: i64,
608 pub last_fetched_epoch: i32,
609 pub log_start_offset: i64,
610 pub max_bytes: i32,
611 pub replica_directory_id: Option<[u8; 16]>,
612 pub high_watermark: Option<i64>,
613}
614
615impl FetchPartitionV18 {
616 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
617 encoder.write_i32(self.partition_index);
618 encoder.write_i32(self.current_leader_epoch);
619 encoder.write_i64(self.fetch_offset);
620 encoder.write_i32(self.last_fetched_epoch);
621 encoder.write_i64(self.log_start_offset);
622 encoder.write_i32(self.max_bytes);
623 write_fetch_partition_v17_tags(
624 encoder,
625 self.replica_directory_id.as_ref(),
626 self.high_watermark,
627 );
628 Ok(())
629 }
630}
631
632pub type FetchForgottenTopicV18 = FetchForgottenTopicV13;
633
634#[derive(Debug, Clone, PartialEq, Eq)]
635pub struct FetchForgottenTopicV13 {
636 pub topic_id: [u8; 16],
637 pub partitions: Vec<i32>,
638}
639
640impl FetchForgottenTopicV13 {
641 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
642 encoder.write_uuid(&self.topic_id);
643 encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
644 encoder.write_i32(*partition);
645 Ok(())
646 })?;
647 encoder.write_empty_tagged_fields();
648 Ok(())
649 }
650}
651
652impl FetchForgottenTopicV12 {
653 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
654 encoder.write_compact_string(&self.name)?;
655 encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
656 encoder.write_i32(*partition);
657 Ok(())
658 })?;
659 encoder.write_empty_tagged_fields();
660 Ok(())
661 }
662}
663
664impl FetchRequestV11 {
665 pub fn encode(&self) -> Result<Vec<u8>> {
666 let mut encoder = Encoder::new();
667 RequestHeader {
668 api_key: API_KEY,
669 api_version: 11,
670 correlation_id: self.correlation_id,
671 client_id: self.client_id.clone(),
672 }
673 .encode_v1(&mut encoder)?;
674 encoder.write_i32(self.replica_id);
675 encoder.write_i32(self.max_wait_ms);
676 encoder.write_i32(self.min_bytes);
677 encoder.write_i32(self.max_bytes);
678 encoder.write_i8(self.isolation_level);
679 encoder.write_i32(self.session_id);
680 encoder.write_i32(self.session_epoch);
681 encoder.write_array(Some(self.topics.as_slice()), |encoder, topic| {
682 topic.encode(encoder)
683 })?;
684 encoder.write_array(Some(self.forgotten_topics.as_slice()), |encoder, topic| {
685 topic.encode(encoder)
686 })?;
687 encoder.write_string(&self.rack_id)?;
688 Ok(encoder.into_bytes())
689 }
690}
691
692#[derive(Debug, Clone, PartialEq, Eq)]
693pub struct FetchTopicV11 {
694 pub name: String,
695 pub partitions: Vec<FetchPartitionV11>,
696}
697
698impl FetchTopicV11 {
699 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
700 encoder.write_string(&self.name)?;
701 encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
702 partition.encode(encoder)
703 })
704 }
705}
706
707#[derive(Debug, Clone, PartialEq, Eq)]
708pub struct FetchPartitionV11 {
709 pub partition_index: i32,
710 pub current_leader_epoch: i32,
711 pub fetch_offset: i64,
712 pub log_start_offset: i64,
713 pub max_bytes: i32,
714}
715
716impl FetchPartitionV11 {
717 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
718 encoder.write_i32(self.partition_index);
719 encoder.write_i32(self.current_leader_epoch);
720 encoder.write_i64(self.fetch_offset);
721 encoder.write_i64(self.log_start_offset);
722 encoder.write_i32(self.max_bytes);
723 Ok(())
724 }
725}
726
727#[derive(Debug, Clone, PartialEq, Eq)]
728pub struct FetchForgottenTopicV11 {
729 pub name: String,
730 pub partitions: Vec<i32>,
731}
732
733impl FetchForgottenTopicV11 {
734 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
735 encoder.write_string(&self.name)?;
736 encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
737 encoder.write_i32(*partition);
738 Ok(())
739 })
740 }
741}
742
743impl FetchRequestV4 {
744 pub fn encode(&self) -> Result<Vec<u8>> {
745 let mut encoder = Encoder::new();
746 RequestHeader {
747 api_key: API_KEY,
748 api_version: 4,
749 correlation_id: self.correlation_id,
750 client_id: self.client_id.clone(),
751 }
752 .encode_v1(&mut encoder)?;
753 encoder.write_i32(self.replica_id);
754 encoder.write_i32(self.max_wait_ms);
755 encoder.write_i32(self.min_bytes);
756 encoder.write_i32(self.max_bytes);
757 encoder.write_i8(self.isolation_level);
758 encoder.write_array(Some(self.topics.as_slice()), |encoder, topic| {
759 topic.encode(encoder)
760 })?;
761 Ok(encoder.into_bytes())
762 }
763}
764
765impl FetchRequestV2 {
766 pub fn encode(&self) -> Result<Vec<u8>> {
767 let mut encoder = Encoder::new();
768 RequestHeader {
769 api_key: API_KEY,
770 api_version: 2,
771 correlation_id: self.correlation_id,
772 client_id: self.client_id.clone(),
773 }
774 .encode_v1(&mut encoder)?;
775 encoder.write_i32(self.replica_id);
776 encoder.write_i32(self.max_wait_ms);
777 encoder.write_i32(self.min_bytes);
778 encoder.write_array(Some(self.topics.as_slice()), |encoder, topic| {
779 topic.encode(encoder)
780 })?;
781 Ok(encoder.into_bytes())
782 }
783}
784
785#[derive(Debug, Clone, PartialEq, Eq)]
786pub struct FetchTopicV2 {
787 pub name: String,
788 pub partitions: Vec<FetchPartitionV2>,
789}
790
791impl FetchTopicV2 {
792 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
793 encoder.write_string(&self.name)?;
794 encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
795 partition.encode(encoder)
796 })
797 }
798}
799
800#[derive(Debug, Clone, PartialEq, Eq)]
801pub struct FetchPartitionV2 {
802 pub partition_index: i32,
803 pub fetch_offset: i64,
804 pub max_bytes: i32,
805}
806
807impl FetchPartitionV2 {
808 fn encode(&self, encoder: &mut Encoder) -> Result<()> {
809 encoder.write_i32(self.partition_index);
810 encoder.write_i64(self.fetch_offset);
811 encoder.write_i32(self.max_bytes);
812 Ok(())
813 }
814}
815
816#[derive(Debug, Clone, PartialEq, Eq)]
817pub struct FetchResponseV2 {
818 pub throttle_time_ms: i32,
819 pub responses: Vec<FetchTopicResponseV2>,
820}
821
822#[derive(Debug, Clone, PartialEq, Eq)]
823pub struct FetchResponseV4 {
824 pub throttle_time_ms: i32,
825 pub responses: Vec<FetchTopicResponseV4>,
826}
827
828#[derive(Debug, Clone, PartialEq, Eq)]
830pub struct FetchResponseV11 {
831 pub throttle_time_ms: i32,
832 pub error_code: i16,
833 pub session_id: i32,
834 pub responses: Vec<FetchTopicResponseV11>,
835}
836
837#[derive(Debug, Clone, PartialEq, Eq)]
839pub struct FetchResponseV12 {
840 pub throttle_time_ms: i32,
841 pub error_code: i16,
842 pub session_id: i32,
843 pub responses: Vec<FetchTopicResponseV12>,
844}
845
846#[derive(Debug, Clone, PartialEq, Eq)]
848pub struct FetchResponseV13 {
849 pub throttle_time_ms: i32,
850 pub error_code: i16,
851 pub session_id: i32,
852 pub responses: Vec<FetchTopicResponseV13>,
853}
854
855impl FetchResponseV12 {
856 pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
857 Ok(Self {
858 throttle_time_ms: decoder.read_i32()?,
859 error_code: decoder.read_i16()?,
860 session_id: decoder.read_i32()?,
861 responses: decoder
862 .read_compact_array("fetch responses", FetchTopicResponseV12::decode)?
863 .unwrap_or_default(),
864 })
865 .and_then(|response| {
866 decoder.read_tagged_fields()?;
867 decoder.finish()?;
868 Ok(response)
869 })
870 }
871}
872
873impl FetchResponseV13 {
874 pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
875 let response = Self {
876 throttle_time_ms: decoder.read_i32()?,
877 error_code: decoder.read_i16()?,
878 session_id: decoder.read_i32()?,
879 responses: decoder
880 .read_compact_array("fetch responses", FetchTopicResponseV13::decode)?
881 .unwrap_or_default(),
882 };
883 decoder.read_tagged_fields()?;
884 decoder.finish()?;
885 Ok(response)
886 }
887}
888
889#[derive(Debug, Clone, PartialEq, Eq)]
890pub struct FetchTopicResponseV13 {
891 pub topic_id: [u8; 16],
892 pub partitions: Vec<FetchPartitionResponseV13>,
893}
894
895impl FetchTopicResponseV13 {
896 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
897 let topic_id = decoder.read_uuid()?;
898 let partitions = decoder
899 .read_compact_array(
900 "fetch partition responses",
901 FetchPartitionResponseV13::decode,
902 )?
903 .unwrap_or_default();
904 decoder.read_tagged_fields()?;
905 Ok(Self {
906 topic_id,
907 partitions,
908 })
909 }
910}
911
912pub type FetchPartitionResponseV13 = FetchPartitionResponseV12;
913pub type FetchResponseV14 = FetchResponseV13;
914pub type FetchTopicResponseV14 = FetchTopicResponseV13;
915pub type FetchPartitionResponseV14 = FetchPartitionResponseV13;
916pub type FetchForgottenTopicV14 = FetchForgottenTopicV13;
917pub type FetchPartitionResponseV16 = FetchPartitionResponseV13;
918pub type FetchTopicResponseV16 = FetchTopicResponseV13;
919pub type FetchResponseV15 = FetchResponseV13;
920pub type FetchTopicResponseV15 = FetchTopicResponseV13;
921pub type FetchPartitionResponseV15 = FetchPartitionResponseV13;
922pub type FetchForgottenTopicV15 = FetchForgottenTopicV13;
923
924#[derive(Debug, Clone, PartialEq, Eq)]
926pub struct FetchNodeEndpointV16 {
927 pub node_id: i32,
928 pub host: String,
929 pub port: i32,
930 pub rack: Option<String>,
931}
932
933#[derive(Debug, Clone, PartialEq, Eq)]
935pub struct FetchResponseV16 {
936 pub throttle_time_ms: i32,
937 pub error_code: i16,
938 pub session_id: i32,
939 pub responses: Vec<FetchTopicResponseV16>,
940 pub node_endpoints: Vec<FetchNodeEndpointV16>,
941}
942
943impl FetchResponseV16 {
944 pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
945 let response = Self {
946 throttle_time_ms: decoder.read_i32()?,
947 error_code: decoder.read_i16()?,
948 session_id: decoder.read_i32()?,
949 responses: decoder
950 .read_compact_array("fetch responses", FetchTopicResponseV16::decode)?
951 .unwrap_or_default(),
952 node_endpoints: Vec::new(),
953 };
954 let tagged_fields = decoder.read_tagged_fields()?;
955 let node_endpoints = tagged_fields
956 .into_iter()
957 .find(|field| field.tag == 0)
958 .map(|field| decode_fetch_node_endpoints(&field.data, decoder.limits()))
959 .transpose()?
960 .unwrap_or_default();
961 Ok(Self {
962 node_endpoints,
963 ..response
964 })
965 }
966}
967
968fn decode_fetch_node_endpoints(
969 data: &[u8],
970 limits: DecodeLimits,
971) -> Result<Vec<FetchNodeEndpointV16>> {
972 let mut decoder = Decoder::with_limits(data, limits);
973 Ok(decoder
974 .read_compact_array("fetch node endpoints", |decoder| {
975 let endpoint = FetchNodeEndpointV16 {
976 node_id: decoder.read_i32()?,
977 host: decoder.read_compact_string()?,
978 port: decoder.read_i32()?,
979 rack: decoder.read_compact_nullable_string()?,
980 };
981 decoder.read_tagged_fields()?;
982 Ok(endpoint)
983 })?
984 .unwrap_or_default())
985}
986
987pub type FetchResponseV17 = FetchResponseV16;
988pub type FetchTopicResponseV17 = FetchTopicResponseV16;
989pub type FetchPartitionResponseV17 = FetchPartitionResponseV16;
990pub type FetchResponseV18 = FetchResponseV17;
991pub type FetchTopicResponseV18 = FetchTopicResponseV17;
992pub type FetchPartitionResponseV18 = FetchPartitionResponseV17;
993
994#[derive(Debug, Clone, PartialEq, Eq)]
995pub struct FetchTopicResponseV12 {
996 pub name: String,
997 pub partitions: Vec<FetchPartitionResponseV12>,
998}
999
1000impl FetchTopicResponseV12 {
1001 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1002 let name = decoder.read_compact_string()?;
1003 let partitions = decoder
1004 .read_compact_array(
1005 "fetch partition responses",
1006 FetchPartitionResponseV12::decode,
1007 )?
1008 .unwrap_or_default();
1009 decoder.read_tagged_fields()?;
1010 Ok(Self { name, partitions })
1011 }
1012}
1013
1014#[derive(Debug, Clone, PartialEq, Eq)]
1015pub struct FetchPartitionResponseV12 {
1016 pub partition_index: i32,
1017 pub error_code: i16,
1018 pub high_watermark: i64,
1019 pub last_stable_offset: i64,
1020 pub log_start_offset: i64,
1021 pub aborted_transactions: Vec<AbortedTransactionV12>,
1022 pub preferred_read_replica: i32,
1023 pub records: Vec<MessageSetRecord>,
1024}
1025
1026impl FetchPartitionResponseV12 {
1027 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1028 let limits = decoder.limits();
1029 let partition_index = decoder.read_i32()?;
1030 let error_code = decoder.read_i16()?;
1031 let high_watermark = decoder.read_i64()?;
1032 let last_stable_offset = decoder.read_i64()?;
1033 let log_start_offset = decoder.read_i64()?;
1034 let aborted_transactions = decoder
1035 .read_compact_array("aborted transactions", AbortedTransactionV12::decode)?
1036 .unwrap_or_default();
1037 let preferred_read_replica = decoder.read_i32()?;
1038 let records = decoder
1039 .read_compact_nullable_bytes()?
1040 .map(|bytes| decode_message_set(&bytes, limits))
1041 .transpose()?
1042 .unwrap_or_default();
1043 decoder.read_tagged_fields()?;
1044 Ok(Self {
1045 partition_index,
1046 error_code,
1047 high_watermark,
1048 last_stable_offset,
1049 log_start_offset,
1050 aborted_transactions,
1051 preferred_read_replica,
1052 records,
1053 })
1054 }
1055}
1056
1057#[derive(Debug, Clone, PartialEq, Eq)]
1058pub struct AbortedTransactionV12 {
1059 pub producer_id: i64,
1060 pub first_offset: i64,
1061}
1062
1063impl AbortedTransactionV12 {
1064 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1065 let producer_id = decoder.read_i64()?;
1066 let first_offset = decoder.read_i64()?;
1067 decoder.read_tagged_fields()?;
1068 Ok(Self {
1069 producer_id,
1070 first_offset,
1071 })
1072 }
1073}
1074
1075impl FetchResponseV11 {
1076 pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
1077 let response = Self {
1078 throttle_time_ms: decoder.read_i32()?,
1079 error_code: decoder.read_i16()?,
1080 session_id: decoder.read_i32()?,
1081 responses: decoder
1082 .read_array("fetch responses", FetchTopicResponseV11::decode)?
1083 .unwrap_or_default(),
1084 };
1085 decoder.finish()?;
1086 Ok(response)
1087 }
1088}
1089
1090#[derive(Debug, Clone, PartialEq, Eq)]
1091pub struct FetchTopicResponseV11 {
1092 pub name: String,
1093 pub partitions: Vec<FetchPartitionResponseV11>,
1094}
1095
1096impl FetchTopicResponseV11 {
1097 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1098 Ok(Self {
1099 name: decoder.read_string()?,
1100 partitions: decoder
1101 .read_array(
1102 "fetch partition responses",
1103 FetchPartitionResponseV11::decode,
1104 )?
1105 .unwrap_or_default(),
1106 })
1107 }
1108}
1109
1110#[derive(Debug, Clone, PartialEq, Eq)]
1111pub struct FetchPartitionResponseV11 {
1112 pub partition_index: i32,
1113 pub error_code: i16,
1114 pub high_watermark: i64,
1115 pub last_stable_offset: i64,
1116 pub log_start_offset: i64,
1117 pub aborted_transactions: Vec<AbortedTransactionV4>,
1118 pub preferred_read_replica: i32,
1119 pub records: Vec<MessageSetRecord>,
1120}
1121
1122impl FetchPartitionResponseV11 {
1123 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1124 let limits = decoder.limits();
1125 Ok(Self {
1126 partition_index: decoder.read_i32()?,
1127 error_code: decoder.read_i16()?,
1128 high_watermark: decoder.read_i64()?,
1129 last_stable_offset: decoder.read_i64()?,
1130 log_start_offset: decoder.read_i64()?,
1131 aborted_transactions: decoder
1132 .read_array("aborted transactions", AbortedTransactionV4::decode)?
1133 .unwrap_or_default(),
1134 preferred_read_replica: decoder.read_i32()?,
1135 records: decode_message_set(&decoder.read_bytes()?, limits)?,
1136 })
1137 }
1138}
1139
1140impl FetchResponseV4 {
1141 pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
1142 let response = Self {
1143 throttle_time_ms: decoder.read_i32()?,
1144 responses: decoder
1145 .read_array("fetch responses", FetchTopicResponseV4::decode)?
1146 .unwrap_or_default(),
1147 };
1148 decoder.finish()?;
1149 Ok(response)
1150 }
1151}
1152
1153#[derive(Debug, Clone, PartialEq, Eq)]
1154pub struct FetchTopicResponseV4 {
1155 pub name: String,
1156 pub partitions: Vec<FetchPartitionResponseV4>,
1157}
1158
1159impl FetchTopicResponseV4 {
1160 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1161 Ok(Self {
1162 name: decoder.read_string()?,
1163 partitions: decoder
1164 .read_array(
1165 "fetch partition responses",
1166 FetchPartitionResponseV4::decode,
1167 )?
1168 .unwrap_or_default(),
1169 })
1170 }
1171}
1172
1173#[derive(Debug, Clone, PartialEq, Eq)]
1174pub struct FetchPartitionResponseV4 {
1175 pub partition_index: i32,
1176 pub error_code: i16,
1177 pub high_watermark: i64,
1178 pub last_stable_offset: i64,
1179 pub aborted_transactions: Vec<AbortedTransactionV4>,
1180 pub records: Vec<MessageSetRecord>,
1181}
1182
1183impl FetchPartitionResponseV4 {
1184 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1185 let limits = decoder.limits();
1186 Ok(Self {
1187 partition_index: decoder.read_i32()?,
1188 error_code: decoder.read_i16()?,
1189 high_watermark: decoder.read_i64()?,
1190 last_stable_offset: decoder.read_i64()?,
1191 aborted_transactions: decoder
1192 .read_array("aborted transactions", AbortedTransactionV4::decode)?
1193 .unwrap_or_default(),
1194 records: decode_message_set(&decoder.read_bytes()?, limits)?,
1195 })
1196 }
1197}
1198
1199#[derive(Debug, Clone, PartialEq, Eq)]
1200pub struct AbortedTransactionV4 {
1201 pub producer_id: i64,
1202 pub first_offset: i64,
1203}
1204
1205impl AbortedTransactionV4 {
1206 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1207 Ok(Self {
1208 producer_id: decoder.read_i64()?,
1209 first_offset: decoder.read_i64()?,
1210 })
1211 }
1212}
1213
1214impl FetchResponseV2 {
1215 pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
1216 Ok(Self {
1217 throttle_time_ms: decoder.read_i32()?,
1218 responses: decoder
1219 .read_array("fetch responses", FetchTopicResponseV2::decode)?
1220 .unwrap_or_default(),
1221 })
1222 }
1223}
1224
1225#[derive(Debug, Clone, PartialEq, Eq)]
1226pub struct FetchTopicResponseV2 {
1227 pub name: String,
1228 pub partitions: Vec<FetchPartitionResponseV2>,
1229}
1230
1231impl FetchTopicResponseV2 {
1232 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1233 Ok(Self {
1234 name: decoder.read_string()?,
1235 partitions: decoder
1236 .read_array(
1237 "fetch partition responses",
1238 FetchPartitionResponseV2::decode,
1239 )?
1240 .unwrap_or_default(),
1241 })
1242 }
1243}
1244
1245#[derive(Debug, Clone, PartialEq, Eq)]
1246pub struct FetchPartitionResponseV2 {
1247 pub partition_index: i32,
1248 pub error_code: i16,
1249 pub high_watermark: i64,
1250 pub records: Vec<MessageSetRecord>,
1251}
1252
1253impl FetchPartitionResponseV2 {
1254 fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
1255 let limits = decoder.limits();
1256 Ok(Self {
1257 partition_index: decoder.read_i32()?,
1258 error_code: decoder.read_i16()?,
1259 high_watermark: decoder.read_i64()?,
1260 records: decode_message_set(&decoder.read_bytes()?, limits)?,
1261 })
1262 }
1263}
1264
1265#[derive(Debug, Clone, PartialEq, Eq)]
1266pub struct MessageSetRecord {
1267 pub offset: i64,
1268 pub leader_epoch: i32,
1271 pub timestamp_ms: i64,
1272 pub key: Option<Vec<u8>>,
1273 pub value: Option<Vec<u8>>,
1274 pub headers: Vec<RecordBatchHeader>,
1275 pub producer_id: Option<i64>,
1276 pub transactional: bool,
1277 pub control: bool,
1278}
1279
1280pub fn decode_message_set(bytes: &[u8], limits: DecodeLimits) -> Result<Vec<MessageSetRecord>> {
1287 let mut decoder = Decoder::with_limits(bytes, limits);
1288 let mut records = Vec::new();
1289
1290 while decoder.remaining() >= 12 {
1291 let offset = decoder.read_i64()?;
1292 let message_size = decoder.read_i32()?;
1293 if message_size < 0 {
1294 return Err(Error::NegativeLength {
1295 kind: "message",
1296 length: message_size,
1297 });
1298 }
1299 let message_size =
1300 usize::try_from(message_size).map_err(|_| Error::LengthOverflow("message"))?;
1301 if decoder.remaining() < message_size {
1303 break;
1304 }
1305 let message = decoder.read_exact(message_size)?;
1306 let decoded = decode_message_or_batch(offset, message, limits)?;
1307 let total = records
1308 .len()
1309 .checked_add(decoded.len())
1310 .ok_or(Error::LengthOverflow("fetch records"))?;
1311 decoder.ensure_collection_length("fetch records", total)?;
1312 records.extend(decoded);
1313 }
1314
1315 Ok(records)
1316}
1317
1318fn decode_message_or_batch(
1319 offset: i64,
1320 bytes: &[u8],
1321 limits: DecodeLimits,
1322) -> Result<Vec<MessageSetRecord>> {
1323 match bytes.get(4).copied() {
1324 Some(2) => decode_record_batch(offset, bytes, limits),
1325 _ => Ok(vec![decode_message(offset, bytes, limits)?]),
1326 }
1327}
1328
1329fn decode_message(offset: i64, bytes: &[u8], limits: DecodeLimits) -> Result<MessageSetRecord> {
1330 let mut decoder = Decoder::with_limits(bytes, limits);
1331 let _crc = decoder.read_i32()?;
1332 let magic = decoder.read_i8()?;
1333 let _attributes = decoder.read_i8()?;
1334 let timestamp_ms = match magic {
1335 0 => -1,
1336 1 => decoder.read_i64()?,
1337 _ => {
1338 return Err(Error::UnsupportedVersion {
1339 kind: "message magic",
1340 version: i16::from(magic),
1341 })
1342 }
1343 };
1344 let key = decoder.read_nullable_bytes()?;
1345 let value = decoder.read_nullable_bytes()?;
1346
1347 Ok(MessageSetRecord {
1348 offset,
1349 leader_epoch: -1,
1350 timestamp_ms,
1351 key,
1352 value,
1353 headers: Vec::new(),
1354 producer_id: None,
1355 transactional: false,
1356 control: false,
1357 })
1358}
1359
1360fn decode_record_batch(
1361 base_offset: i64,
1362 bytes: &[u8],
1363 limits: DecodeLimits,
1364) -> Result<Vec<MessageSetRecord>> {
1365 let mut decoder = Decoder::with_limits(bytes, limits);
1366 let partition_leader_epoch = decoder.read_i32()?;
1367 let magic = decoder.read_i8()?;
1368 if magic != 2 {
1369 return Err(Error::UnsupportedVersion {
1370 kind: "record batch magic",
1371 version: i16::from(magic),
1372 });
1373 }
1374 let _crc = decoder.read_i32()?;
1375 let attributes = decoder.read_i16()?;
1376 let compression = RecordBatchCompression::from_attributes(attributes)?;
1377 let _last_offset_delta = decoder.read_i32()?;
1378 let base_timestamp = decoder.read_i64()?;
1379 let _max_timestamp = decoder.read_i64()?;
1380 let producer_id = decoder.read_i64()?;
1381 let _producer_epoch = decoder.read_i16()?;
1382 let _base_sequence = decoder.read_i32()?;
1383 let record_count = decoder.read_i32()?;
1384 if record_count < 0 {
1385 return Err(Error::NegativeLength {
1386 kind: "record batch records",
1387 length: record_count,
1388 });
1389 }
1390
1391 let record_count =
1392 usize::try_from(record_count).map_err(|_| Error::LengthOverflow("record batch records"))?;
1393 decoder.ensure_collection_length("record batch records", record_count)?;
1394 let record_bytes = if compression.is_compressed() {
1395 let compressed = decoder.read_exact(decoder.remaining())?;
1396 decompress_record_batch_records_with_limit(
1397 compression,
1398 compressed,
1399 limits.max_decompressed_record_bytes(),
1400 )?
1401 } else {
1402 if decoder.remaining() > limits.max_decompressed_record_bytes() {
1403 return Err(Error::LimitExceeded {
1404 kind: "decompressed record batch bytes",
1405 actual: decoder.remaining(),
1406 max: limits.max_decompressed_record_bytes(),
1407 });
1408 }
1409 decoder.read_exact(decoder.remaining())?.to_vec()
1410 };
1411 let mut record_decoder = Decoder::with_limits(&record_bytes, limits);
1412 let mut records = Vec::with_capacity(record_count);
1413 for _ in 0..record_count {
1414 let record_length = record_decoder.read_varint()?;
1415 if record_length < 0 {
1416 return Err(Error::NegativeLength {
1417 kind: "record",
1418 length: record_length,
1419 });
1420 }
1421 let record_length =
1422 usize::try_from(record_length).map_err(|_| Error::LengthOverflow("record"))?;
1423 let record_bytes = record_decoder.read_exact(record_length)?;
1424 records.push(decode_record(
1425 base_offset,
1426 partition_leader_epoch,
1427 base_timestamp,
1428 producer_id,
1429 attributes,
1430 record_bytes,
1431 limits,
1432 )?);
1433 }
1434
1435 Ok(records)
1436}
1437
1438fn decode_record(
1439 base_offset: i64,
1440 partition_leader_epoch: i32,
1441 base_timestamp: i64,
1442 producer_id: i64,
1443 batch_attributes: i16,
1444 bytes: &[u8],
1445 limits: DecodeLimits,
1446) -> Result<MessageSetRecord> {
1447 let mut decoder = Decoder::with_limits(bytes, limits);
1448 let _attributes = decoder.read_i8()?;
1449 let timestamp_delta = decoder.read_varlong()?;
1450 let offset_delta = decoder.read_varint()?;
1451 let key = decoder.read_varint_nullable_bytes()?;
1452 let value = decoder.read_varint_nullable_bytes()?;
1453 let header_count = decoder.read_varint()?;
1454 if header_count < 0 {
1455 return Err(Error::NegativeLength {
1456 kind: "record headers",
1457 length: header_count,
1458 });
1459 }
1460 let header_count =
1461 usize::try_from(header_count).map_err(|_| Error::LengthOverflow("record headers"))?;
1462 decoder.ensure_collection_length("record headers", header_count)?;
1463 let mut headers = Vec::with_capacity(header_count);
1464 for _ in 0..header_count {
1465 let header_key =
1466 String::from_utf8(decoder.read_varint_bytes()?).map_err(|_| Error::InvalidUtf8)?;
1467 let header_value = decoder.read_varint_nullable_bytes()?;
1468 headers.push(RecordBatchHeader::new(header_key, header_value));
1469 }
1470
1471 Ok(MessageSetRecord {
1472 offset: base_offset.saturating_add(i64::from(offset_delta)),
1473 leader_epoch: partition_leader_epoch,
1474 timestamp_ms: base_timestamp.saturating_add(timestamp_delta),
1475 key,
1476 value,
1477 headers,
1478 producer_id: (producer_id >= 0).then_some(producer_id),
1479 transactional: batch_attributes & 0x10 != 0,
1480 control: batch_attributes & 0x20 != 0,
1481 })
1482}
1483
1484#[cfg(test)]
1485#[allow(clippy::unwrap_used)]
1486mod tests {
1487 use super::{
1488 FetchPartitionV11, FetchPartitionV12, FetchPartitionV17, FetchPartitionV18,
1489 FetchPartitionV2, FetchReplicaStateV15, FetchRequestV11, FetchRequestV12, FetchRequestV13,
1490 FetchRequestV14, FetchRequestV15, FetchRequestV16, FetchRequestV17, FetchRequestV18,
1491 FetchRequestV2, FetchRequestV4, FetchResponseV11, FetchResponseV12, FetchResponseV13,
1492 FetchResponseV16, FetchResponseV2, FetchResponseV4, FetchTopicV11, FetchTopicV12,
1493 FetchTopicV13, FetchTopicV17, FetchTopicV18, FetchTopicV2, MessageSetRecord,
1494 RecordBatchHeader,
1495 };
1496 use crate::codec::{Decoder, Encoder};
1497
1498 #[test]
1499 fn encodes_fetch_request_v2() {
1500 let request = FetchRequestV2 {
1501 correlation_id: 7,
1502 client_id: Some("kafrust".to_owned()),
1503 replica_id: -1,
1504 max_wait_ms: 500,
1505 min_bytes: 1,
1506 topics: vec![FetchTopicV2 {
1507 name: "orders".to_owned(),
1508 partitions: vec![FetchPartitionV2 {
1509 partition_index: 0,
1510 fetch_offset: 42,
1511 max_bytes: 1_048_576,
1512 }],
1513 }],
1514 };
1515
1516 let bytes = request.encode().unwrap();
1517 assert_eq!(&bytes[0..4], &[0, 1, 0, 2]);
1518 assert!(bytes.len() > 40);
1519 }
1520
1521 #[test]
1522 fn encodes_fetch_request_v4() {
1523 let request = FetchRequestV4 {
1524 correlation_id: 8,
1525 client_id: Some("kafrust".to_owned()),
1526 replica_id: -1,
1527 max_wait_ms: 500,
1528 min_bytes: 1,
1529 max_bytes: 1_048_576,
1530 isolation_level: 0,
1531 topics: vec![FetchTopicV2 {
1532 name: "orders".to_owned(),
1533 partitions: vec![FetchPartitionV2 {
1534 partition_index: 0,
1535 fetch_offset: 42,
1536 max_bytes: 1_048_576,
1537 }],
1538 }],
1539 };
1540
1541 let bytes = request.encode().unwrap();
1542 assert_eq!(&bytes[0..4], &[0, 1, 0, 4]);
1543 assert_eq!(&bytes[4..8], &[0, 0, 0, 8]);
1544 assert!(bytes.len() > 45);
1545 }
1546
1547 #[test]
1548 fn encodes_fetch_request_v11_with_rack_and_fetch_session_fields() {
1549 let request = FetchRequestV11 {
1550 correlation_id: 9,
1551 client_id: Some("kafrust".to_owned()),
1552 replica_id: -1,
1553 max_wait_ms: 500,
1554 min_bytes: 1,
1555 max_bytes: 1_048_576,
1556 isolation_level: 1,
1557 session_id: 0,
1558 session_epoch: 0,
1559 topics: vec![FetchTopicV11 {
1560 name: "orders".to_owned(),
1561 partitions: vec![FetchPartitionV11 {
1562 partition_index: 0,
1563 current_leader_epoch: -1,
1564 fetch_offset: 42,
1565 log_start_offset: -1,
1566 max_bytes: 1_048_576,
1567 }],
1568 }],
1569 forgotten_topics: Vec::new(),
1570 rack_id: "rack-a".to_owned(),
1571 };
1572
1573 let bytes = request.encode().unwrap();
1574 let mut decoder = Decoder::new(&bytes);
1575 assert_eq!(decoder.read_i16().unwrap(), 1);
1576 assert_eq!(decoder.read_i16().unwrap(), 11);
1577 assert_eq!(decoder.read_i32().unwrap(), 9);
1578 assert_eq!(
1579 decoder.read_nullable_string().unwrap().as_deref(),
1580 Some("kafrust")
1581 );
1582 assert_eq!(decoder.read_i32().unwrap(), -1);
1583 assert_eq!(decoder.read_i32().unwrap(), 500);
1584 assert_eq!(decoder.read_i32().unwrap(), 1);
1585 assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1586 assert_eq!(decoder.read_i8().unwrap(), 1);
1587 assert_eq!(decoder.read_i32().unwrap(), 0);
1588 assert_eq!(decoder.read_i32().unwrap(), 0);
1589 assert_eq!(decoder.read_i32().unwrap(), 1);
1590 assert_eq!(decoder.read_string().unwrap(), "orders");
1591 assert_eq!(decoder.read_i32().unwrap(), 1);
1592 assert_eq!(decoder.read_i32().unwrap(), 0);
1593 assert_eq!(decoder.read_i32().unwrap(), -1);
1594 assert_eq!(decoder.read_i64().unwrap(), 42);
1595 assert_eq!(decoder.read_i64().unwrap(), -1);
1596 assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1597 assert_eq!(decoder.read_i32().unwrap(), 0);
1598 assert_eq!(decoder.read_string().unwrap(), "rack-a");
1599 assert!(decoder.is_empty());
1600 }
1601
1602 #[test]
1603 fn decodes_fetch_response_v11_with_preferred_read_replica() {
1604 let mut bytes = Encoder::new();
1605 bytes.write_i32(3);
1606 bytes.write_i16(0);
1607 bytes.write_i32(17);
1608 bytes.write_i32(1);
1609 bytes.write_string("orders").unwrap();
1610 bytes.write_i32(1);
1611 bytes.write_i32(0);
1612 bytes.write_i16(0);
1613 bytes.write_i64(43);
1614 bytes.write_i64(42);
1615 bytes.write_i64(40);
1616 bytes.write_i32(0);
1617 bytes.write_i32(2);
1618 bytes.write_bytes(&[]).unwrap();
1619
1620 let bytes = bytes.into_bytes();
1621 let mut decoder = Decoder::new(&bytes);
1622 let response = FetchResponseV11::decode_body(&mut decoder).unwrap();
1623 let partition = &response.responses[0].partitions[0];
1624
1625 assert_eq!(response.throttle_time_ms, 3);
1626 assert_eq!(response.session_id, 17);
1627 assert_eq!(partition.log_start_offset, 40);
1628 assert_eq!(partition.preferred_read_replica, 2);
1629 assert!(partition.records.is_empty());
1630 assert!(decoder.is_empty());
1631 }
1632
1633 #[test]
1634 fn encodes_fetch_request_v12_with_flexible_rack_fields() {
1635 let request = FetchRequestV12 {
1636 correlation_id: 10,
1637 client_id: Some("kafrust".to_owned()),
1638 replica_id: -1,
1639 max_wait_ms: 500,
1640 min_bytes: 1,
1641 max_bytes: 1_048_576,
1642 isolation_level: 1,
1643 session_id: 0,
1644 session_epoch: 0,
1645 topics: vec![FetchTopicV12 {
1646 name: "orders".to_owned(),
1647 partitions: vec![FetchPartitionV12 {
1648 partition_index: 0,
1649 current_leader_epoch: -1,
1650 fetch_offset: 42,
1651 last_fetched_epoch: -1,
1652 log_start_offset: -1,
1653 max_bytes: 1_048_576,
1654 }],
1655 }],
1656 forgotten_topics: Vec::new(),
1657 rack_id: "rack-a".to_owned(),
1658 };
1659
1660 let bytes = request.encode().unwrap();
1661 let mut decoder = Decoder::new(&bytes);
1662 assert_eq!(decoder.read_i16().unwrap(), 1);
1663 assert_eq!(decoder.read_i16().unwrap(), 12);
1664 assert_eq!(decoder.read_i32().unwrap(), 10);
1665 assert_eq!(
1666 decoder.read_nullable_string().unwrap().as_deref(),
1667 Some("kafrust")
1668 );
1669 assert_eq!(decoder.read_unsigned_varint().unwrap(), 0);
1670 assert_eq!(decoder.read_i32().unwrap(), -1);
1671 assert_eq!(decoder.read_i32().unwrap(), 500);
1672 assert_eq!(decoder.read_i32().unwrap(), 1);
1673 assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1674 assert_eq!(decoder.read_i8().unwrap(), 1);
1675 assert_eq!(decoder.read_i32().unwrap(), 0);
1676 assert_eq!(decoder.read_i32().unwrap(), 0);
1677 assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
1678 assert_eq!(decoder.read_compact_string().unwrap(), "orders");
1679 assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
1680 assert_eq!(decoder.read_i32().unwrap(), 0);
1681 assert_eq!(decoder.read_i32().unwrap(), -1);
1682 assert_eq!(decoder.read_i64().unwrap(), 42);
1683 assert_eq!(decoder.read_i32().unwrap(), -1);
1684 assert_eq!(decoder.read_i64().unwrap(), -1);
1685 assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1686 assert_eq!(decoder.read_unsigned_varint().unwrap(), 0);
1687 assert_eq!(decoder.read_unsigned_varint().unwrap(), 0);
1688 assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
1689 assert_eq!(decoder.read_compact_string().unwrap(), "rack-a");
1690 assert_eq!(decoder.read_unsigned_varint().unwrap(), 0);
1691 assert!(decoder.is_empty());
1692 }
1693
1694 #[test]
1695 fn decodes_fetch_response_v12_with_preferred_read_replica() {
1696 let mut bytes = Encoder::new();
1697 bytes.write_i32(3);
1698 bytes.write_i16(0);
1699 bytes.write_i32(17);
1700 bytes.write_unsigned_varint(2);
1701 bytes.write_compact_string("orders").unwrap();
1702 bytes.write_unsigned_varint(2);
1703 bytes.write_i32(0);
1704 bytes.write_i16(0);
1705 bytes.write_i64(43);
1706 bytes.write_i64(42);
1707 bytes.write_i64(40);
1708 bytes.write_unsigned_varint(2);
1709 bytes.write_i64(7);
1710 bytes.write_i64(40);
1711 bytes.write_unsigned_varint(0);
1712 bytes.write_i32(2);
1713 bytes.write_compact_nullable_bytes(Some(&[])).unwrap();
1714 bytes.write_unsigned_varint(0);
1715 bytes.write_unsigned_varint(0);
1716 bytes.write_unsigned_varint(0);
1717
1718 let bytes = bytes.into_bytes();
1719 let mut decoder = Decoder::new(&bytes);
1720 let response = FetchResponseV12::decode_body(&mut decoder).unwrap();
1721 let partition = &response.responses[0].partitions[0];
1722
1723 assert_eq!(response.throttle_time_ms, 3);
1724 assert_eq!(response.session_id, 17);
1725 assert_eq!(partition.log_start_offset, 40);
1726 assert_eq!(partition.preferred_read_replica, 2);
1727 assert_eq!(partition.aborted_transactions[0].producer_id, 7);
1728 assert!(partition.records.is_empty());
1729 assert!(decoder.is_empty());
1730 }
1731
1732 #[test]
1733 fn encodes_fetch_request_v13_with_topic_uuid_and_cluster_tag() {
1734 let request = FetchRequestV13 {
1735 correlation_id: 11,
1736 client_id: Some("kafrust".to_owned()),
1737 cluster_id: Some("cluster-a".to_owned()),
1738 replica_id: -1,
1739 max_wait_ms: 500,
1740 min_bytes: 1,
1741 max_bytes: 1_048_576,
1742 isolation_level: 1,
1743 session_id: 17,
1744 session_epoch: 2,
1745 topics: vec![FetchTopicV13 {
1746 topic_id: [3; 16],
1747 partitions: vec![FetchPartitionV12 {
1748 partition_index: 0,
1749 current_leader_epoch: 4,
1750 fetch_offset: 42,
1751 last_fetched_epoch: 3,
1752 log_start_offset: -1,
1753 max_bytes: 1_048_576,
1754 }],
1755 }],
1756 forgotten_topics: Vec::new(),
1757 rack_id: "rack-a".to_owned(),
1758 };
1759
1760 let bytes = request.encode().unwrap();
1761 let mut decoder = Decoder::new(&bytes);
1762 assert_eq!(decoder.read_i16().unwrap(), 1);
1763 assert_eq!(decoder.read_i16().unwrap(), 13);
1764 assert_eq!(decoder.read_i32().unwrap(), 11);
1765 assert_eq!(
1766 decoder.read_nullable_string().unwrap().as_deref(),
1767 Some("kafrust")
1768 );
1769 assert!(decoder.read_tagged_fields().unwrap().is_empty());
1770 assert_eq!(decoder.read_i32().unwrap(), -1);
1771 assert_eq!(decoder.read_i32().unwrap(), 500);
1772 assert_eq!(decoder.read_i32().unwrap(), 1);
1773 assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1774 assert_eq!(decoder.read_i8().unwrap(), 1);
1775 assert_eq!(decoder.read_i32().unwrap(), 17);
1776 assert_eq!(decoder.read_i32().unwrap(), 2);
1777 assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
1778 assert_eq!(decoder.read_uuid().unwrap(), [3; 16]);
1779 assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
1780 assert_eq!(decoder.read_i32().unwrap(), 0);
1781 assert_eq!(decoder.read_i32().unwrap(), 4);
1782 assert_eq!(decoder.read_i64().unwrap(), 42);
1783 assert_eq!(decoder.read_i32().unwrap(), 3);
1784 assert_eq!(decoder.read_i64().unwrap(), -1);
1785 assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1786 assert!(decoder.read_tagged_fields().unwrap().is_empty());
1787 assert!(decoder.read_tagged_fields().unwrap().is_empty());
1788 assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
1789 assert_eq!(decoder.read_compact_string().unwrap(), "rack-a");
1790 let fields = decoder.read_tagged_fields().unwrap();
1791 assert_eq!(fields.len(), 1);
1792 let mut cluster_decoder = Decoder::new(&fields[0].data);
1793 assert_eq!(fields[0].tag, 0);
1794 assert_eq!(cluster_decoder.read_compact_string().unwrap(), "cluster-a");
1795 assert!(cluster_decoder.is_empty());
1796 assert!(decoder.is_empty());
1797 }
1798
1799 #[test]
1800 fn decodes_fetch_response_v13_with_topic_uuid() {
1801 let mut bytes = Encoder::new();
1802 bytes.write_i32(3);
1803 bytes.write_i16(0);
1804 bytes.write_i32(17);
1805 bytes.write_unsigned_varint(2);
1806 bytes.write_uuid(&[4; 16]);
1807 bytes.write_unsigned_varint(2);
1808 bytes.write_i32(0);
1809 bytes.write_i16(0);
1810 bytes.write_i64(43);
1811 bytes.write_i64(42);
1812 bytes.write_i64(40);
1813 bytes.write_unsigned_varint(1);
1814 bytes.write_i32(2);
1815 bytes.write_compact_nullable_bytes(Some(&[])).unwrap();
1816 bytes.write_empty_tagged_fields();
1817 bytes.write_empty_tagged_fields();
1818 bytes.write_empty_tagged_fields();
1819
1820 let bytes = bytes.into_bytes();
1821 let mut decoder = Decoder::new(&bytes);
1822 let response = FetchResponseV13::decode_body(&mut decoder).unwrap();
1823 let partition = &response.responses[0].partitions[0];
1824
1825 assert_eq!(response.session_id, 17);
1826 assert_eq!(response.responses[0].topic_id, [4; 16]);
1827 assert_eq!(partition.high_watermark, 43);
1828 assert_eq!(partition.preferred_read_replica, 2);
1829 assert!(partition.records.is_empty());
1830 assert!(decoder.is_empty());
1831 }
1832
1833 #[test]
1834 fn decodes_fetch_response_v16_node_endpoints_tag() {
1835 let mut endpoint = Encoder::new();
1836 endpoint.write_unsigned_varint(2);
1837 endpoint.write_i32(3);
1838 endpoint.write_compact_string("broker-a").unwrap();
1839 endpoint.write_i32(9092);
1840 endpoint
1841 .write_compact_nullable_string(Some("rack-a"))
1842 .unwrap();
1843 endpoint.write_empty_tagged_fields();
1844 let endpoint = endpoint.into_bytes();
1845
1846 let mut bytes = Encoder::new();
1847 bytes.write_i32(3);
1848 bytes.write_i16(0);
1849 bytes.write_i32(17);
1850 bytes.write_unsigned_varint(1);
1851 bytes.write_unsigned_varint(1);
1852 bytes.write_unsigned_varint(0);
1853 bytes.write_unsigned_varint(endpoint.len() as u32);
1854 bytes.write_raw(&endpoint);
1855
1856 let bytes = bytes.into_bytes();
1857 let mut decoder = Decoder::new(&bytes);
1858 let response = FetchResponseV16::decode_body(&mut decoder).unwrap();
1859 assert_eq!(response.node_endpoints.len(), 1);
1860 assert_eq!(response.node_endpoints[0].node_id, 3);
1861 assert_eq!(response.node_endpoints[0].host, "broker-a");
1862 assert_eq!(response.node_endpoints[0].port, 9092);
1863 assert_eq!(response.node_endpoints[0].rack.as_deref(), Some("rack-a"));
1864 assert!(decoder.is_empty());
1865 }
1866
1867 #[test]
1868 fn encodes_fetch_request_v14_with_the_v14_header_version() {
1869 let request = FetchRequestV14 {
1870 correlation_id: 12,
1871 client_id: None,
1872 cluster_id: None,
1873 replica_id: -1,
1874 max_wait_ms: 500,
1875 min_bytes: 1,
1876 max_bytes: 1_048_576,
1877 isolation_level: 0,
1878 session_id: 0,
1879 session_epoch: -1,
1880 topics: Vec::new(),
1881 forgotten_topics: Vec::new(),
1882 rack_id: String::new(),
1883 };
1884
1885 let bytes = request.encode().unwrap();
1886 assert_eq!(&bytes[0..4], &[0, 1, 0, 14]);
1887 let mut decoder = Decoder::new(&bytes);
1888 decoder.read_i16().unwrap();
1889 decoder.read_i16().unwrap();
1890 decoder.read_i32().unwrap();
1891 decoder.read_nullable_string().unwrap();
1892 assert!(decoder.read_tagged_fields().unwrap().is_empty());
1893 assert_eq!(decoder.read_i32().unwrap(), -1);
1894 assert_eq!(decoder.read_i32().unwrap(), 500);
1895 assert_eq!(decoder.read_i32().unwrap(), 1);
1896 assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1897 assert_eq!(decoder.read_i8().unwrap(), 0);
1898 assert_eq!(decoder.read_i32().unwrap(), 0);
1899 assert_eq!(decoder.read_i32().unwrap(), -1);
1900 assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
1901 assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
1902 assert!(decoder.read_compact_string().unwrap().is_empty());
1903 assert!(decoder.read_tagged_fields().unwrap().is_empty());
1904 assert!(decoder.is_empty());
1905 }
1906
1907 #[test]
1908 fn encodes_fetch_request_v15_replica_state_as_tagged_struct() {
1909 let request = FetchRequestV15 {
1910 correlation_id: 13,
1911 client_id: Some("kafrust".to_owned()),
1912 cluster_id: Some("cluster-a".to_owned()),
1913 replica_state: Some(FetchReplicaStateV15 {
1914 replica_id: 4,
1915 replica_epoch: 9,
1916 }),
1917 max_wait_ms: 500,
1918 min_bytes: 1,
1919 max_bytes: 1_048_576,
1920 isolation_level: 0,
1921 session_id: 0,
1922 session_epoch: -1,
1923 topics: Vec::new(),
1924 forgotten_topics: Vec::new(),
1925 rack_id: String::new(),
1926 };
1927
1928 let bytes = request.encode().unwrap();
1929 let mut decoder = Decoder::new(&bytes);
1930 assert_eq!(decoder.read_i16().unwrap(), 1);
1931 assert_eq!(decoder.read_i16().unwrap(), 15);
1932 assert_eq!(decoder.read_i32().unwrap(), 13);
1933 assert_eq!(
1934 decoder.read_nullable_string().unwrap().as_deref(),
1935 Some("kafrust")
1936 );
1937 assert!(decoder.read_tagged_fields().unwrap().is_empty());
1938 assert_eq!(decoder.read_i32().unwrap(), 500);
1939 assert_eq!(decoder.read_i32().unwrap(), 1);
1940 assert_eq!(decoder.read_i32().unwrap(), 1_048_576);
1941 assert_eq!(decoder.read_i8().unwrap(), 0);
1942 assert_eq!(decoder.read_i32().unwrap(), 0);
1943 assert_eq!(decoder.read_i32().unwrap(), -1);
1944 assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
1945 assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
1946 assert!(decoder.read_compact_string().unwrap().is_empty());
1947 let fields = decoder.read_tagged_fields().unwrap();
1948 assert_eq!(fields.len(), 2);
1949 assert_eq!(fields[0].tag, 0);
1950 assert_eq!(fields[1].tag, 1);
1951 let mut cluster_decoder = Decoder::new(&fields[0].data);
1952 assert_eq!(cluster_decoder.read_compact_string().unwrap(), "cluster-a");
1953 let mut replica_decoder = Decoder::new(&fields[1].data);
1954 assert_eq!(replica_decoder.read_i32().unwrap(), 4);
1955 assert_eq!(replica_decoder.read_i64().unwrap(), 9);
1956 assert!(decoder.is_empty());
1957 }
1958
1959 #[test]
1960 fn encodes_fetch_request_v16_with_the_v16_header_version() {
1961 let request = FetchRequestV16 {
1962 correlation_id: 14,
1963 client_id: None,
1964 cluster_id: None,
1965 replica_state: None,
1966 max_wait_ms: 500,
1967 min_bytes: 1,
1968 max_bytes: 1_048_576,
1969 isolation_level: 0,
1970 session_id: 0,
1971 session_epoch: -1,
1972 topics: Vec::new(),
1973 forgotten_topics: Vec::new(),
1974 rack_id: String::new(),
1975 };
1976
1977 let bytes = request.encode().unwrap();
1978 assert_eq!(&bytes[0..4], &[0, 1, 0, 16]);
1979 }
1980
1981 #[test]
1982 fn encodes_fetch_request_v17_directory_id_as_partition_tag() {
1983 let request = FetchRequestV17 {
1984 correlation_id: 15,
1985 client_id: None,
1986 cluster_id: None,
1987 replica_state: None,
1988 max_wait_ms: 500,
1989 min_bytes: 1,
1990 max_bytes: 1_048_576,
1991 isolation_level: 0,
1992 session_id: 0,
1993 session_epoch: -1,
1994 topics: vec![FetchTopicV17 {
1995 topic_id: [7; 16],
1996 partitions: vec![FetchPartitionV17 {
1997 partition_index: 0,
1998 current_leader_epoch: -1,
1999 fetch_offset: 42,
2000 last_fetched_epoch: -1,
2001 log_start_offset: -1,
2002 max_bytes: 1_048_576,
2003 replica_directory_id: Some([8; 16]),
2004 }],
2005 }],
2006 forgotten_topics: Vec::new(),
2007 rack_id: String::new(),
2008 };
2009
2010 let bytes = request.encode().unwrap();
2011 let mut decoder = Decoder::new(&bytes);
2012 assert_eq!(decoder.read_i16().unwrap(), 1);
2013 assert_eq!(decoder.read_i16().unwrap(), 17);
2014 decoder.read_i32().unwrap();
2015 decoder.read_nullable_string().unwrap();
2016 assert!(decoder.read_tagged_fields().unwrap().is_empty());
2017 for _ in 0..3 {
2018 decoder.read_i32().unwrap();
2019 }
2020 decoder.read_i8().unwrap();
2021 decoder.read_i32().unwrap();
2022 decoder.read_i32().unwrap();
2023 assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
2024 assert_eq!(decoder.read_uuid().unwrap(), [7; 16]);
2025 assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
2026 decoder.read_i32().unwrap();
2027 decoder.read_i32().unwrap();
2028 decoder.read_i64().unwrap();
2029 decoder.read_i32().unwrap();
2030 decoder.read_i64().unwrap();
2031 decoder.read_i32().unwrap();
2032 let fields = decoder.read_tagged_fields().unwrap();
2033 assert_eq!(fields.len(), 1);
2034 assert_eq!(fields[0].tag, 0);
2035 assert_eq!(fields[0].data, vec![8; 16]);
2036 assert!(decoder.read_tagged_fields().unwrap().is_empty());
2037 assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
2038 assert!(decoder.read_compact_string().unwrap().is_empty());
2039 assert!(decoder.read_tagged_fields().unwrap().is_empty());
2040 assert!(decoder.is_empty());
2041 }
2042
2043 #[test]
2044 fn encodes_fetch_request_v18_high_watermark_after_directory_id() {
2045 let request = FetchRequestV18 {
2046 correlation_id: 16,
2047 client_id: None,
2048 cluster_id: None,
2049 replica_state: None,
2050 max_wait_ms: 500,
2051 min_bytes: 1,
2052 max_bytes: 1_048_576,
2053 isolation_level: 0,
2054 session_id: 0,
2055 session_epoch: -1,
2056 topics: vec![FetchTopicV18 {
2057 topic_id: [9; 16],
2058 partitions: vec![FetchPartitionV18 {
2059 partition_index: 0,
2060 current_leader_epoch: -1,
2061 fetch_offset: 42,
2062 last_fetched_epoch: -1,
2063 log_start_offset: -1,
2064 max_bytes: 1_048_576,
2065 replica_directory_id: Some([10; 16]),
2066 high_watermark: Some(100),
2067 }],
2068 }],
2069 forgotten_topics: Vec::new(),
2070 rack_id: String::new(),
2071 };
2072
2073 let bytes = request.encode().unwrap();
2074 let mut decoder = Decoder::new(&bytes);
2075 decoder.read_i16().unwrap();
2076 decoder.read_i16().unwrap();
2077 decoder.read_i32().unwrap();
2078 decoder.read_nullable_string().unwrap();
2079 decoder.read_tagged_fields().unwrap();
2080 for _ in 0..3 {
2081 decoder.read_i32().unwrap();
2082 }
2083 decoder.read_i8().unwrap();
2084 decoder.read_i32().unwrap();
2085 decoder.read_i32().unwrap();
2086 decoder.read_unsigned_varint().unwrap();
2087 decoder.read_uuid().unwrap();
2088 decoder.read_unsigned_varint().unwrap();
2089 decoder.read_i32().unwrap();
2090 decoder.read_i32().unwrap();
2091 decoder.read_i64().unwrap();
2092 decoder.read_i32().unwrap();
2093 decoder.read_i64().unwrap();
2094 decoder.read_i32().unwrap();
2095 let fields = decoder.read_tagged_fields().unwrap();
2096 assert_eq!(fields.len(), 2);
2097 assert_eq!(fields[0].tag, 0);
2098 assert_eq!(fields[0].data, vec![10; 16]);
2099 assert_eq!(fields[1].tag, 1);
2100 assert_eq!(fields[1].data, 100_i64.to_be_bytes());
2101 assert!(decoder.read_tagged_fields().unwrap().is_empty());
2102 assert_eq!(decoder.read_unsigned_varint().unwrap(), 1);
2103 assert!(decoder.read_compact_string().unwrap().is_empty());
2104 assert!(decoder.read_tagged_fields().unwrap().is_empty());
2105 assert!(decoder.is_empty());
2106 }
2107
2108 #[test]
2109 fn decodes_fetch_response_v4_with_aborted_transaction() {
2110 let mut bytes = Encoder::new();
2111 bytes.write_i32(0);
2112 bytes.write_i32(1);
2113 bytes.write_string("orders").unwrap();
2114 bytes.write_i32(1);
2115 bytes.write_i32(0);
2116 bytes.write_i16(0);
2117 bytes.write_i64(43);
2118 bytes.write_i64(42);
2119 bytes.write_i32(1);
2120 bytes.write_i64(7);
2121 bytes.write_i64(40);
2122 bytes.write_bytes(&[]).unwrap();
2123 let bytes = bytes.into_bytes();
2124
2125 let mut decoder = Decoder::new(&bytes);
2126 let response = FetchResponseV4::decode_body(&mut decoder).unwrap();
2127 let partition = &response.responses[0].partitions[0];
2128
2129 assert_eq!(partition.high_watermark, 43);
2130 assert_eq!(partition.last_stable_offset, 42);
2131 assert_eq!(partition.aborted_transactions[0].producer_id, 7);
2132 assert_eq!(partition.aborted_transactions[0].first_offset, 40);
2133 assert!(partition.records.is_empty());
2134 assert!(decoder.is_empty());
2135 }
2136
2137 #[test]
2138 fn decodes_fetch_response_v2_with_message_set() {
2139 let mut message = Encoder::new();
2140 message.write_i32(0);
2141 message.write_i8(1);
2142 message.write_i8(0);
2143 message.write_i64(123);
2144 message.write_nullable_bytes(Some(b"order-1")).unwrap();
2145 message.write_nullable_bytes(Some(b"created")).unwrap();
2146 let message = message.into_bytes();
2147
2148 let mut set = Encoder::new();
2149 set.write_i64(42);
2150 set.write_i32(i32::try_from(message.len()).unwrap());
2151 set.write_raw(&message);
2152 let set = set.into_bytes();
2153
2154 let mut bytes = Encoder::new();
2155 bytes.write_i32(0);
2156 bytes.write_i32(1);
2157 bytes.write_string("orders").unwrap();
2158 bytes.write_i32(1);
2159 bytes.write_i32(0);
2160 bytes.write_i16(0);
2161 bytes.write_i64(43);
2162 bytes.write_bytes(&set).unwrap();
2163 let bytes = bytes.into_bytes();
2164
2165 let mut decoder = Decoder::new(&bytes);
2166 let response = FetchResponseV2::decode_body(&mut decoder).unwrap();
2167 let record = MessageSetRecord {
2168 offset: 42,
2169 leader_epoch: -1,
2170 timestamp_ms: 123,
2171 key: Some(b"order-1".to_vec()),
2172 value: Some(b"created".to_vec()),
2173 headers: Vec::new(),
2174 producer_id: None,
2175 transactional: false,
2176 control: false,
2177 };
2178
2179 assert_eq!(response.throttle_time_ms, 0);
2180 assert_eq!(response.responses[0].partitions[0].high_watermark, 43);
2181 assert_eq!(response.responses[0].partitions[0].records, vec![record]);
2182 assert!(decoder.is_empty());
2183 }
2184
2185 #[test]
2186 fn decodes_fetch_response_v2_ignores_partial_trailing_message_set_entry() {
2187 let mut message = Encoder::new();
2188 message.write_i32(0);
2189 message.write_i8(1);
2190 message.write_i8(0);
2191 message.write_i64(123);
2192 message.write_nullable_bytes(Some(b"order-1")).unwrap();
2193 message.write_nullable_bytes(Some(b"created")).unwrap();
2194 let message = message.into_bytes();
2195
2196 let mut set = Encoder::new();
2197 set.write_i64(42);
2198 set.write_i32(i32::try_from(message.len()).unwrap());
2199 set.write_raw(&message);
2200 set.write_i64(-1);
2201 set.write_i32(61);
2202 set.write_raw(&[0; 22]);
2203 let set = set.into_bytes();
2204
2205 let mut bytes = Encoder::new();
2206 bytes.write_i32(0);
2207 bytes.write_i32(1);
2208 bytes.write_string("orders").unwrap();
2209 bytes.write_i32(1);
2210 bytes.write_i32(0);
2211 bytes.write_i16(0);
2212 bytes.write_i64(43);
2213 bytes.write_bytes(&set).unwrap();
2214 let bytes = bytes.into_bytes();
2215
2216 let mut decoder = Decoder::new(&bytes);
2217 let response = FetchResponseV2::decode_body(&mut decoder).unwrap();
2218
2219 assert_eq!(response.responses[0].partitions[0].records.len(), 1);
2220 assert_eq!(response.responses[0].partitions[0].records[0].offset, 42);
2221 assert!(decoder.is_empty());
2222 }
2223
2224 #[test]
2225 fn decodes_fetch_response_v2_with_record_batch() {
2226 let mut record = Vec::new();
2227 record.push(0);
2228 write_varlong(&mut record, 5);
2229 write_varint(&mut record, 0);
2230 write_varint(&mut record, 7);
2231 record.extend_from_slice(b"order-1");
2232 write_varint(&mut record, 7);
2233 record.extend_from_slice(b"created");
2234 write_varint(&mut record, 2);
2235 write_varint(&mut record, 6);
2236 record.extend_from_slice(b"source");
2237 write_varint(&mut record, 8);
2238 record.extend_from_slice(b"checkout");
2239 write_varint(&mut record, 9);
2240 record.extend_from_slice(b"tombstone");
2241 write_varint(&mut record, -1);
2242
2243 let mut batch = Encoder::new();
2244 batch.write_i32(0);
2245 batch.write_i8(2);
2246 batch.write_i32(0);
2247 batch.write_i16(0x10);
2248 batch.write_i32(0);
2249 batch.write_i64(1_000);
2250 batch.write_i64(1_005);
2251 batch.write_i64(7);
2252 batch.write_i16(-1);
2253 batch.write_i32(-1);
2254 batch.write_i32(1);
2255 let mut encoded_record = Vec::new();
2256 write_varint(&mut encoded_record, i32::try_from(record.len()).unwrap());
2257 encoded_record.extend_from_slice(&record);
2258 batch.write_raw(&encoded_record);
2259 let batch = batch.into_bytes();
2260
2261 let mut set = Encoder::new();
2262 set.write_i64(42);
2263 set.write_i32(i32::try_from(batch.len()).unwrap());
2264 set.write_raw(&batch);
2265 let set = set.into_bytes();
2266
2267 let mut bytes = Encoder::new();
2268 bytes.write_i32(0);
2269 bytes.write_i32(1);
2270 bytes.write_string("orders").unwrap();
2271 bytes.write_i32(1);
2272 bytes.write_i32(0);
2273 bytes.write_i16(0);
2274 bytes.write_i64(43);
2275 bytes.write_bytes(&set).unwrap();
2276 let bytes = bytes.into_bytes();
2277
2278 let mut decoder = Decoder::new(&bytes);
2279 let response = FetchResponseV2::decode_body(&mut decoder).unwrap();
2280 let record = MessageSetRecord {
2281 offset: 42,
2282 leader_epoch: 0,
2283 timestamp_ms: 1_005,
2284 key: Some(b"order-1".to_vec()),
2285 value: Some(b"created".to_vec()),
2286 headers: vec![
2287 RecordBatchHeader::new("source", Some(b"checkout".to_vec())),
2288 RecordBatchHeader::new("tombstone", None),
2289 ],
2290 producer_id: Some(7),
2291 transactional: true,
2292 control: false,
2293 };
2294
2295 assert_eq!(response.responses[0].partitions[0].records, vec![record]);
2296 assert!(decoder.is_empty());
2297 }
2298
2299 fn write_varint(output: &mut Vec<u8>, value: i32) {
2300 write_unsigned_varint(output, u64::from(((value << 1) ^ (value >> 31)) as u32));
2301 }
2302
2303 fn write_varlong(output: &mut Vec<u8>, value: i64) {
2304 write_unsigned_varint(output, ((value << 1) ^ (value >> 63)) as u64);
2305 }
2306
2307 fn write_unsigned_varint(output: &mut Vec<u8>, mut value: u64) {
2308 loop {
2309 let mut byte = (value & 0x7f) as u8;
2310 value >>= 7;
2311 if value != 0 {
2312 byte |= 0x80;
2313 }
2314 output.push(byte);
2315 if value == 0 {
2316 break;
2317 }
2318 }
2319 }
2320}