1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 45;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct AlterPartitionReassignmentsRequestV0 {
9 pub correlation_id: i32,
10 pub client_id: Option<String>,
11 pub timeout_ms: i32,
12 pub topics: Vec<AlterPartitionReassignmentsTopicV0>,
13}
14
15impl AlterPartitionReassignmentsRequestV0 {
16 pub fn encode(&self) -> Result<Vec<u8>> {
17 let mut encoder = Encoder::new();
18 RequestHeader {
19 api_key: API_KEY,
20 api_version: 0,
21 correlation_id: self.correlation_id,
22 client_id: self.client_id.clone(),
23 }
24 .encode_v2(&mut encoder)?;
25 encoder.write_i32(self.timeout_ms);
26 encoder.write_compact_array(Some(&self.topics), |encoder, topic| {
27 encoder.write_compact_string(&topic.name)?;
28 encoder.write_compact_array(Some(&topic.partitions), |encoder, partition| {
29 encoder.write_i32(partition.partition_index);
30 encoder.write_compact_array(
31 partition.replicas.as_deref(),
32 |encoder, replica| {
33 encoder.write_i32(*replica);
34 Ok(())
35 },
36 )?;
37 encoder.write_empty_tagged_fields();
38 Ok(())
39 })?;
40 encoder.write_empty_tagged_fields();
41 Ok(())
42 })?;
43 encoder.write_empty_tagged_fields();
44 Ok(encoder.into_bytes())
45 }
46}
47
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct AlterPartitionReassignmentsTopicV0 {
50 pub name: String,
51 pub partitions: Vec<AlterPartitionReassignmentsPartitionV0>,
52}
53
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct AlterPartitionReassignmentsPartitionV0 {
56 pub partition_index: i32,
57 pub replicas: Option<Vec<i32>>,
58}
59
60#[derive(Debug, Clone, PartialEq, Eq)]
61pub struct AlterPartitionReassignmentsResponseV0 {
62 pub throttle_time_ms: i32,
63 pub error_code: i16,
64 pub error_message: Option<String>,
65 pub responses: Vec<AlterPartitionReassignmentsTopicResponseV0>,
66}
67
68impl AlterPartitionReassignmentsResponseV0 {
69 pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
70 let throttle_time_ms = decoder.read_i32()?;
71 let error_code = decoder.read_i16()?;
72 let error_message = decoder.read_compact_nullable_string()?;
73 let responses = decoder
74 .read_compact_array("alter partition reassignment responses", |decoder| {
75 let name = decoder.read_compact_string()?;
76 let partitions = decoder
77 .read_compact_array(
78 "alter partition reassignment partition responses",
79 |decoder| {
80 let partition_index = decoder.read_i32()?;
81 let error_code = decoder.read_i16()?;
82 let error_message = decoder.read_compact_nullable_string()?;
83 decoder.read_tagged_fields()?;
84 Ok(AlterPartitionReassignmentsPartitionResponseV0 {
85 partition_index,
86 error_code,
87 error_message,
88 })
89 },
90 )?
91 .unwrap_or_default();
92 decoder.read_tagged_fields()?;
93 Ok(AlterPartitionReassignmentsTopicResponseV0 { name, partitions })
94 })?
95 .unwrap_or_default();
96 decoder.read_tagged_fields()?;
97 Ok(Self {
98 throttle_time_ms,
99 error_code,
100 error_message,
101 responses,
102 })
103 }
104}
105
106#[derive(Debug, Clone, PartialEq, Eq)]
107pub struct AlterPartitionReassignmentsTopicResponseV0 {
108 pub name: String,
109 pub partitions: Vec<AlterPartitionReassignmentsPartitionResponseV0>,
110}
111
112#[derive(Debug, Clone, PartialEq, Eq)]
113pub struct AlterPartitionReassignmentsPartitionResponseV0 {
114 pub partition_index: i32,
115 pub error_code: i16,
116 pub error_message: Option<String>,
117}
118
119#[cfg(test)]
120#[allow(clippy::unwrap_used)]
121mod tests {
122 use super::{
123 AlterPartitionReassignmentsPartitionV0, AlterPartitionReassignmentsRequestV0,
124 AlterPartitionReassignmentsResponseV0, AlterPartitionReassignmentsTopicV0, API_KEY,
125 };
126 use crate::codec::{Decoder, Encoder};
127
128 #[test]
129 fn encodes_alter_partition_reassignments_v0_request() {
130 let request = AlterPartitionReassignmentsRequestV0 {
131 correlation_id: 12,
132 client_id: None,
133 timeout_ms: 30_000,
134 topics: vec![AlterPartitionReassignmentsTopicV0 {
135 name: "orders".to_owned(),
136 partitions: vec![
137 AlterPartitionReassignmentsPartitionV0 {
138 partition_index: 0,
139 replicas: Some(vec![3, 1, 2]),
140 },
141 AlterPartitionReassignmentsPartitionV0 {
142 partition_index: 1,
143 replicas: None,
144 },
145 ],
146 }],
147 };
148
149 let bytes = request.encode().unwrap();
150 assert_eq!(&bytes[0..4], &[0, API_KEY as u8, 0, 0]);
151 assert_eq!(&bytes[4..8], &[0, 0, 0, 12]);
152 assert!(bytes.windows(4).any(|window| window == [0, 0, 0, 3]));
153 assert_eq!(bytes[bytes.len() - 1], 0);
154 }
155
156 #[test]
157 fn decodes_alter_partition_reassignments_v0_response() {
158 let mut bytes = Encoder::new();
159 bytes.write_i32(7);
160 bytes.write_i16(0);
161 bytes.write_compact_nullable_string(None).unwrap();
162 bytes.write_unsigned_varint(2);
163 bytes.write_compact_string("orders").unwrap();
164 bytes.write_unsigned_varint(2);
165 bytes.write_i32(0);
166 bytes.write_i16(0);
167 bytes.write_compact_nullable_string(None).unwrap();
168 bytes.write_empty_tagged_fields();
169 bytes.write_empty_tagged_fields();
170 bytes.write_empty_tagged_fields();
171 let bytes = bytes.into_bytes();
172 let mut decoder = Decoder::new(&bytes);
173
174 let response = AlterPartitionReassignmentsResponseV0::decode_body(&mut decoder).unwrap();
175
176 assert_eq!(response.throttle_time_ms, 7);
177 assert_eq!(response.responses[0].name, "orders");
178 assert_eq!(response.responses[0].partitions[0].partition_index, 0);
179 assert_eq!(response.responses[0].partitions[0].error_code, 0);
180 assert!(decoder.is_empty());
181 }
182}