krabka_protocol/legacy_compat/
fetch.rs

1use crate::{
2    kafka_3_6_2,
3    owned::{
4        fetch_request::{self, FetchRequest},
5        fetch_response::{self, FetchResponse},
6    },
7};
8
9// ── Request: legacy → canonical ──────────────────────────────────────────────
10
11impl From<kafka_3_6_2::owned::fetch_request::FetchRequest> for FetchRequest {
12    fn from(l: kafka_3_6_2::owned::fetch_request::FetchRequest) -> Self {
13        Self {
14            replica_id: l.replica_id,
15            max_wait_ms: l.max_wait_ms,
16            min_bytes: l.min_bytes,
17            max_bytes: l.max_bytes,
18            isolation_level: l.isolation_level,
19            session_id: l.session_id,
20            session_epoch: l.session_epoch,
21            topics: l.topics.into_iter().map(Into::into).collect(),
22            forgotten_topics_data: l
23                .forgotten_topics_data
24                .into_iter()
25                .map(Into::into)
26                .collect(),
27            rack_id: l.rack_id,
28            cluster_id: l.cluster_id,
29            replica_state: l.replica_state.into(),
30            ..Default::default()
31        }
32    }
33}
34
35impl From<kafka_3_6_2::owned::fetch_request::ReplicaState> for fetch_request::ReplicaState {
36    fn from(l: kafka_3_6_2::owned::fetch_request::ReplicaState) -> Self {
37        Self {
38            replica_id: l.replica_id,
39            replica_epoch: l.replica_epoch,
40            ..Default::default()
41        }
42    }
43}
44
45impl From<kafka_3_6_2::owned::fetch_request::FetchTopic> for fetch_request::FetchTopic {
46    fn from(l: kafka_3_6_2::owned::fetch_request::FetchTopic) -> Self {
47        Self {
48            topic: l.topic,
49            // topic_id (v13+) defaults to Uuid::nil()
50            partitions: l.partitions.into_iter().map(Into::into).collect(),
51            ..Default::default()
52        }
53    }
54}
55
56impl From<kafka_3_6_2::owned::fetch_request::FetchPartition> for fetch_request::FetchPartition {
57    fn from(l: kafka_3_6_2::owned::fetch_request::FetchPartition) -> Self {
58        Self {
59            partition: l.partition,
60            current_leader_epoch: l.current_leader_epoch,
61            fetch_offset: l.fetch_offset,
62            last_fetched_epoch: l.last_fetched_epoch,
63            log_start_offset: l.log_start_offset,
64            partition_max_bytes: l.partition_max_bytes,
65            // replica_directory_id (v17+ tagged) and high_watermark (v18+ tagged) default
66            ..Default::default()
67        }
68    }
69}
70
71impl From<kafka_3_6_2::owned::fetch_request::ForgottenTopic> for fetch_request::ForgottenTopic {
72    fn from(l: kafka_3_6_2::owned::fetch_request::ForgottenTopic) -> Self {
73        Self {
74            topic: l.topic,
75            // topic_id (v13+) defaults to Uuid::nil()
76            partitions: l.partitions,
77            ..Default::default()
78        }
79    }
80}
81
82// ── Response: canonical → legacy ─────────────────────────────────────────────
83
84impl From<FetchResponse> for kafka_3_6_2::owned::fetch_response::FetchResponse {
85    fn from(c: FetchResponse) -> Self {
86        Self {
87            throttle_time_ms: c.throttle_time_ms,
88            error_code: c.error_code,
89            session_id: c.session_id,
90            responses: c.responses.into_iter().map(Into::into).collect(),
91            // node_endpoints (canonical v16+ tagged field) dropped — not present in 3.6.2 schema
92            ..Default::default()
93        }
94    }
95}
96
97impl From<fetch_response::FetchableTopicResponse>
98    for kafka_3_6_2::owned::fetch_response::FetchableTopicResponse
99{
100    fn from(c: fetch_response::FetchableTopicResponse) -> Self {
101        Self {
102            topic: c.topic,
103            topic_id: c.topic_id,
104            partitions: c.partitions.into_iter().map(Into::into).collect(),
105            ..Default::default()
106        }
107    }
108}
109
110impl From<fetch_response::PartitionData> for kafka_3_6_2::owned::fetch_response::PartitionData {
111    fn from(c: fetch_response::PartitionData) -> Self {
112        Self {
113            partition_index: c.partition_index,
114            error_code: c.error_code,
115            high_watermark: c.high_watermark,
116            last_stable_offset: c.last_stable_offset,
117            log_start_offset: c.log_start_offset,
118            aborted_transactions: c
119                .aborted_transactions
120                .map(|v| v.into_iter().map(Into::into).collect()),
121            preferred_read_replica: c.preferred_read_replica,
122            records: c.records,
123            diverging_epoch: c.diverging_epoch.into(),
124            current_leader: c.current_leader.into(),
125            snapshot_id: c.snapshot_id.into(),
126            ..Default::default()
127        }
128    }
129}
130
131impl From<fetch_response::AbortedTransaction>
132    for kafka_3_6_2::owned::fetch_response::AbortedTransaction
133{
134    fn from(c: fetch_response::AbortedTransaction) -> Self {
135        Self {
136            producer_id: c.producer_id,
137            first_offset: c.first_offset,
138            ..Default::default()
139        }
140    }
141}
142
143impl From<fetch_response::EpochEndOffset> for kafka_3_6_2::owned::fetch_response::EpochEndOffset {
144    fn from(c: fetch_response::EpochEndOffset) -> Self {
145        Self {
146            epoch: c.epoch,
147            end_offset: c.end_offset,
148            ..Default::default()
149        }
150    }
151}
152
153impl From<fetch_response::LeaderIdAndEpoch>
154    for kafka_3_6_2::owned::fetch_response::LeaderIdAndEpoch
155{
156    fn from(c: fetch_response::LeaderIdAndEpoch) -> Self {
157        Self {
158            leader_id: c.leader_id,
159            leader_epoch: c.leader_epoch,
160            ..Default::default()
161        }
162    }
163}
164
165impl From<fetch_response::SnapshotId> for kafka_3_6_2::owned::fetch_response::SnapshotId {
166    fn from(c: fetch_response::SnapshotId) -> Self {
167        Self {
168            end_offset: c.end_offset,
169            epoch: c.epoch,
170            ..Default::default()
171        }
172    }
173}
174
175#[cfg(test)]
176mod tests {
177
178    use bytes::Bytes;
179
180    use super::*;
181    use crate::{UnknownTaggedFields, primitives::uuid::Uuid, records::RecordsPayload};
182
183    #[test]
184    fn legacy_fetch_request_conversion_preserves_mapped_fields() {
185        let mut legacy = kafka_3_6_2::owned::fetch_request::FetchRequest::populated(15);
186        legacy.replica_id = 42;
187        legacy.max_wait_ms = 43;
188        legacy.min_bytes = 44;
189        legacy.max_bytes = 45;
190        legacy.isolation_level = 1;
191        legacy.session_id = 46;
192        legacy.session_epoch = 47;
193        legacy.rack_id = "rack-a".into();
194        legacy.cluster_id = Some("cluster-a".into());
195        legacy.replica_state.replica_id = 48;
196        legacy.replica_state.replica_epoch = 49;
197        legacy.topics[0].topic = "fetch-topic".into();
198        legacy.topics[0].partitions[0].partition = 3;
199        legacy.topics[0].partitions[0].current_leader_epoch = 4;
200        legacy.topics[0].partitions[0].fetch_offset = 5;
201        legacy.topics[0].partitions[0].last_fetched_epoch = 6;
202        legacy.topics[0].partitions[0].log_start_offset = 7;
203        legacy.topics[0].partitions[0].partition_max_bytes = 8;
204        legacy.forgotten_topics_data[0].topic = "forgotten-topic".into();
205        legacy.forgotten_topics_data[0].partitions = vec![9, 10];
206        let converted = FetchRequest::from(legacy);
207
208        let expected = FetchRequest {
209            replica_id: 42,
210            max_wait_ms: 43,
211            min_bytes: 44,
212            max_bytes: 45,
213            isolation_level: 1,
214            session_id: 46,
215            session_epoch: 47,
216            topics: vec![fetch_request::FetchTopic {
217                topic: "fetch-topic".to_owned(),
218                topic_id: Uuid::ZERO,
219                partitions: vec![fetch_request::FetchPartition {
220                    partition: 3,
221                    current_leader_epoch: 4,
222                    fetch_offset: 5,
223                    last_fetched_epoch: 6,
224                    log_start_offset: 7,
225                    partition_max_bytes: 8,
226                    replica_directory_id: Uuid::ZERO,
227                    high_watermark: 9_223_372_036_854_775_807,
228                    unknown_tagged_fields: UnknownTaggedFields(vec![]),
229                }],
230                unknown_tagged_fields: UnknownTaggedFields(vec![]),
231            }],
232            forgotten_topics_data: vec![fetch_request::ForgottenTopic {
233                topic: "forgotten-topic".to_owned(),
234                topic_id: Uuid::ZERO,
235                partitions: vec![9, 10],
236                unknown_tagged_fields: UnknownTaggedFields(vec![]),
237            }],
238            rack_id: "rack-a".to_owned(),
239            cluster_id: Some("cluster-a".to_owned()),
240            replica_state: fetch_request::ReplicaState {
241                replica_id: 48,
242                replica_epoch: 49,
243                unknown_tagged_fields: UnknownTaggedFields(vec![]),
244            },
245            unknown_tagged_fields: UnknownTaggedFields(vec![]),
246        };
247        assert2::assert!(converted == expected);
248    }
249
250    #[test]
251    fn fetch_response_conversion_preserves_mapped_fields() {
252        let mut canonical = FetchResponse::populated(18);
253        canonical.throttle_time_ms = 51;
254        canonical.error_code = 52;
255        canonical.session_id = 53;
256        canonical.responses[0].topic = "fetch-response-topic".into();
257        canonical.responses[0].topic_id = Uuid([0x12; 16]);
258        canonical.responses[0].partitions[0].partition_index = 54;
259        canonical.responses[0].partitions[0].error_code = 55;
260        canonical.responses[0].partitions[0].high_watermark = 56;
261        canonical.responses[0].partitions[0].last_stable_offset = 57;
262        canonical.responses[0].partitions[0].log_start_offset = 58;
263        canonical.responses[0].partitions[0].preferred_read_replica = 59;
264        canonical.responses[0].partitions[0].records =
265            Some(RecordsPayload::Legacy(Bytes::from_static(&[1, 2, 3])));
266        canonical.responses[0].partitions[0].diverging_epoch.epoch = 60;
267        canonical.responses[0].partitions[0]
268            .diverging_epoch
269            .end_offset = 61;
270        canonical.responses[0].partitions[0]
271            .current_leader
272            .leader_id = 62;
273        canonical.responses[0].partitions[0]
274            .current_leader
275            .leader_epoch = 63;
276        canonical.responses[0].partitions[0].snapshot_id.end_offset = 64;
277        canonical.responses[0].partitions[0].snapshot_id.epoch = 65;
278        canonical.responses[0].partitions[0]
279            .aborted_transactions
280            .as_mut()
281            .expect("aborted transactions")[0]
282            .producer_id = 66;
283        canonical.responses[0].partitions[0]
284            .aborted_transactions
285            .as_mut()
286            .expect("aborted transactions")[0]
287            .first_offset = 67;
288        let converted = kafka_3_6_2::owned::fetch_response::FetchResponse::from(canonical);
289
290        let expected = kafka_3_6_2::owned::fetch_response::FetchResponse {
291            throttle_time_ms: 51,
292            error_code: 52,
293            session_id: 53,
294            responses: vec![kafka_3_6_2::owned::fetch_response::FetchableTopicResponse {
295                topic: "fetch-response-topic".to_owned(),
296                topic_id: Uuid([0x12; 16]),
297                partitions: vec![kafka_3_6_2::owned::fetch_response::PartitionData {
298                    partition_index: 54,
299                    error_code: 55,
300                    high_watermark: 56,
301                    last_stable_offset: 57,
302                    log_start_offset: 58,
303                    aborted_transactions: Some(vec![
304                        kafka_3_6_2::owned::fetch_response::AbortedTransaction {
305                            producer_id: 66,
306                            first_offset: 67,
307                            unknown_tagged_fields: UnknownTaggedFields(vec![]),
308                        },
309                    ]),
310                    preferred_read_replica: 59,
311                    records: Some(RecordsPayload::Legacy(Bytes::from_static(&[1, 2, 3]))),
312                    diverging_epoch: kafka_3_6_2::owned::fetch_response::EpochEndOffset {
313                        epoch: 60,
314                        end_offset: 61,
315                        unknown_tagged_fields: UnknownTaggedFields(vec![]),
316                    },
317                    current_leader: kafka_3_6_2::owned::fetch_response::LeaderIdAndEpoch {
318                        leader_id: 62,
319                        leader_epoch: 63,
320                        unknown_tagged_fields: UnknownTaggedFields(vec![]),
321                    },
322                    snapshot_id: kafka_3_6_2::owned::fetch_response::SnapshotId {
323                        end_offset: 64,
324                        epoch: 65,
325                        unknown_tagged_fields: UnknownTaggedFields(vec![]),
326                    },
327                    unknown_tagged_fields: UnknownTaggedFields(vec![]),
328                }],
329                unknown_tagged_fields: UnknownTaggedFields(vec![]),
330            }],
331            unknown_tagged_fields: UnknownTaggedFields(vec![]),
332        };
333        assert2::assert!(converted == expected);
334    }
335}
krabka_protocol/legacy_compat/fetch.rs

krabka_protocol/legacy_compat/
fetch.rs
