krabka_protocol/legacy_compat/
produce.rs

1use crate::{
2    kafka_3_6_2,
3    owned::{produce_request::ProduceRequest, produce_response::ProduceResponse},
4};
5
6// ── Request: legacy → canonical ──────────────────────────────────────────────
7
8impl From<kafka_3_6_2::owned::produce_request::ProduceRequest> for ProduceRequest {
9    fn from(legacy: kafka_3_6_2::owned::produce_request::ProduceRequest) -> Self {
10        Self {
11            transactional_id: legacy.transactional_id,
12            acks: legacy.acks,
13            timeout_ms: legacy.timeout_ms,
14            topic_data: legacy.topic_data.into_iter().map(Into::into).collect(),
15            ..Default::default()
16        }
17    }
18}
19
20impl From<kafka_3_6_2::owned::produce_request::TopicProduceData>
21    for crate::owned::produce_request::TopicProduceData
22{
23    fn from(l: kafka_3_6_2::owned::produce_request::TopicProduceData) -> Self {
24        Self {
25            name: l.name,
26            partition_data: l.partition_data.into_iter().map(Into::into).collect(),
27            // topic_id (v13+) defaults to Uuid::nil()
28            ..Default::default()
29        }
30    }
31}
32
33impl From<kafka_3_6_2::owned::produce_request::PartitionProduceData>
34    for crate::owned::produce_request::PartitionProduceData
35{
36    fn from(l: kafka_3_6_2::owned::produce_request::PartitionProduceData) -> Self {
37        Self {
38            index: l.index,
39            // RecordsPayload is the same type across both namespaces
40            records: l.records,
41            ..Default::default()
42        }
43    }
44}
45
46// ── Response: canonical → legacy ─────────────────────────────────────────────
47
48impl From<ProduceResponse> for kafka_3_6_2::owned::produce_response::ProduceResponse {
49    fn from(c: ProduceResponse) -> Self {
50        Self {
51            responses: c.responses.into_iter().map(Into::into).collect(),
52            throttle_time_ms: c.throttle_time_ms,
53            ..Default::default()
54        }
55    }
56}
57
58impl From<crate::owned::produce_response::TopicProduceResponse>
59    for kafka_3_6_2::owned::produce_response::TopicProduceResponse
60{
61    fn from(c: crate::owned::produce_response::TopicProduceResponse) -> Self {
62        Self {
63            name: c.name,
64            partition_responses: c.partition_responses.into_iter().map(Into::into).collect(),
65            ..Default::default()
66        }
67    }
68}
69
70impl From<crate::owned::produce_response::PartitionProduceResponse>
71    for kafka_3_6_2::owned::produce_response::PartitionProduceResponse
72{
73    fn from(c: crate::owned::produce_response::PartitionProduceResponse) -> Self {
74        Self {
75            index: c.index,
76            error_code: c.error_code,
77            base_offset: c.base_offset,
78            log_append_time_ms: c.log_append_time_ms,
79            log_start_offset: c.log_start_offset,
80            record_errors: c.record_errors.into_iter().map(Into::into).collect(),
81            error_message: c.error_message,
82            ..Default::default() // Defaulted: current_leader (canonical-only v10+ tagged field)
83        }
84    }
85}
86
87impl From<crate::owned::produce_response::BatchIndexAndErrorMessage>
88    for kafka_3_6_2::owned::produce_response::BatchIndexAndErrorMessage
89{
90    fn from(c: crate::owned::produce_response::BatchIndexAndErrorMessage) -> Self {
91        Self {
92            batch_index: c.batch_index,
93            batch_index_error_message: c.batch_index_error_message,
94            ..Default::default()
95        }
96    }
97}
98
99#[cfg(test)]
100mod tests {
101
102    use bytes::Bytes;
103
104    use super::*;
105    use crate::{UnknownTaggedFields, records::RecordsPayload};
106
107    #[test]
108    fn legacy_produce_request_conversion_preserves_mapped_fields() {
109        let mut legacy = kafka_3_6_2::owned::produce_request::ProduceRequest::populated(9);
110        legacy.transactional_id = Some("txn-a".into());
111        legacy.acks = 1;
112        legacy.timeout_ms = 72;
113        legacy.topic_data[0].name = "produce-topic".into();
114        legacy.topic_data[0].partition_data[0].index = 73;
115        legacy.topic_data[0].partition_data[0].records =
116            Some(RecordsPayload::Legacy(Bytes::from_static(&[4, 5, 6])));
117        let converted = ProduceRequest::from(legacy);
118
119        let expected = ProduceRequest {
120            transactional_id: Some("txn-a".to_string()),
121            acks: 1,
122            timeout_ms: 72,
123            topic_data: vec![crate::owned::produce_request::TopicProduceData {
124                name: "produce-topic".to_string(),
125                topic_id: crate::primitives::uuid::Uuid::ZERO,
126                partition_data: vec![crate::owned::produce_request::PartitionProduceData {
127                    index: 73,
128                    records: Some(RecordsPayload::Legacy(Bytes::from_static(&[4, 5, 6]))),
129                    unknown_tagged_fields: UnknownTaggedFields(vec![]),
130                }],
131                unknown_tagged_fields: UnknownTaggedFields(vec![]),
132            }],
133            unknown_tagged_fields: UnknownTaggedFields(vec![]),
134        };
135        assert2::assert!(converted == expected);
136    }
137
138    #[test]
139    fn produce_response_conversion_preserves_mapped_fields() {
140        let mut canonical = ProduceResponse::populated(12);
141        canonical.throttle_time_ms = 81;
142        canonical.responses[0].name = "produce-response-topic".into();
143        canonical.responses[0].partition_responses[0].index = 82;
144        canonical.responses[0].partition_responses[0].error_code = 83;
145        canonical.responses[0].partition_responses[0].base_offset = 84;
146        canonical.responses[0].partition_responses[0].log_append_time_ms = 85;
147        canonical.responses[0].partition_responses[0].log_start_offset = 86;
148        canonical.responses[0].partition_responses[0].error_message = Some("produce-error".into());
149        canonical.responses[0].partition_responses[0].record_errors[0].batch_index = 87;
150        canonical.responses[0].partition_responses[0].record_errors[0].batch_index_error_message =
151            Some("batch-error".into());
152        let converted = kafka_3_6_2::owned::produce_response::ProduceResponse::from(canonical);
153
154        let expected = kafka_3_6_2::owned::produce_response::ProduceResponse {
155            responses: vec![kafka_3_6_2::owned::produce_response::TopicProduceResponse {
156                name: "produce-response-topic".to_string(),
157                partition_responses: vec![
158                    kafka_3_6_2::owned::produce_response::PartitionProduceResponse {
159                        index: 82,
160                        error_code: 83,
161                        base_offset: 84,
162                        log_append_time_ms: 85,
163                        log_start_offset: 86,
164                        record_errors: vec![
165                            kafka_3_6_2::owned::produce_response::BatchIndexAndErrorMessage {
166                                batch_index: 87,
167                                batch_index_error_message: Some("batch-error".to_string()),
168                                unknown_tagged_fields: UnknownTaggedFields(vec![]),
169                            },
170                        ],
171                        error_message: Some("produce-error".to_string()),
172                        unknown_tagged_fields: UnknownTaggedFields(vec![]),
173                    },
174                ],
175                unknown_tagged_fields: UnknownTaggedFields(vec![]),
176            }],
177            throttle_time_ms: 81,
178            unknown_tagged_fields: UnknownTaggedFields(vec![]),
179        };
180        assert2::assert!(converted == expected);
181    }
182}
krabka_protocol/legacy_compat/produce.rs

krabka_protocol/legacy_compat/
produce.rs
