1use bytes::BufMut;
3
4use crate::{
5 DecodeBorrow, Encode, ProtocolError, UnknownTaggedFields,
6 owned::streams_group_heartbeat_request,
7 primitives::{
8 array::{
9 array_len_prefix_len, get_array_len, get_nullable_array_len,
10 nullable_array_len_prefix_len, put_array_len, put_nullable_array_len,
11 },
12 fixed::{get_bool, get_i8, get_i16, get_i32, put_bool, put_i16, put_i32},
13 string_bytes::{
14 compact_nullable_string_len, compact_string_len, nullable_string_len,
15 put_compact_nullable_string, put_compact_string, put_nullable_string, put_string,
16 string_len,
17 },
18 string_bytes_borrowed::{
19 get_compact_nullable_string_borrowed, get_compact_string_borrowed,
20 get_nullable_string_borrowed, get_string_borrowed,
21 },
22 },
23 tagged_fields::{WriteTaggedFields, read_tagged_fields, tagged_fields_len},
24};
25pub const API_KEY: i16 = 88;
26pub const MIN_VERSION: i16 = 0;
27pub const MAX_VERSION: i16 = 1;
28pub const LATEST_STABLE_VERSION: i16 = 1;
29pub const FLEXIBLE_MIN: i16 = 0;
30#[inline]
31#[must_use]
32pub fn is_flexible(version: i16) -> bool {
33 version >= FLEXIBLE_MIN
34}
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct StreamsGroupHeartbeatRequest<'a> {
37 pub group_id: &'a str,
38 pub member_id: &'a str,
39 pub member_epoch: i32,
40 pub endpoint_information_epoch: i32,
41 pub instance_id: Option<&'a str>,
42 pub rack_id: Option<&'a str>,
43 pub rebalance_timeout_ms: i32,
44 pub topology: Option<Topology<'a>>,
45 pub active_tasks:
46 Option<Vec<super::common::streams_group_heartbeat_request::task_ids::TaskIds<'a>>>,
47 pub standby_tasks:
48 Option<Vec<super::common::streams_group_heartbeat_request::task_ids::TaskIds<'a>>>,
49 pub warmup_tasks:
50 Option<Vec<super::common::streams_group_heartbeat_request::task_ids::TaskIds<'a>>>,
51 pub process_id: Option<&'a str>,
52 pub user_endpoint:
53 Option<super::common::streams_group_heartbeat_request::endpoint::Endpoint<'a>>,
54 pub client_tags:
55 Option<Vec<super::common::streams_group_heartbeat_request::key_value::KeyValue<'a>>>,
56 pub task_offsets:
57 Option<Vec<super::common::streams_group_heartbeat_request::task_offset::TaskOffset<'a>>>,
58 pub task_end_offsets:
59 Option<Vec<super::common::streams_group_heartbeat_request::task_offset::TaskOffset<'a>>>,
60 pub shutdown_application: bool,
61 pub unknown_tagged_fields: UnknownTaggedFields,
62}
63impl Default for StreamsGroupHeartbeatRequest<'_> {
64 fn default() -> Self {
65 Self {
66 group_id: "",
67 member_id: "",
68 member_epoch: 0i32,
69 endpoint_information_epoch: 0i32,
70 instance_id: None,
71 rack_id: None,
72 rebalance_timeout_ms: -1i32,
73 topology: None,
74 active_tasks: None,
75 standby_tasks: None,
76 warmup_tasks: None,
77 process_id: None,
78 user_endpoint: None,
79 client_tags: None,
80 task_offsets: None,
81 task_end_offsets: None,
82 shutdown_application: false,
83 unknown_tagged_fields: UnknownTaggedFields::default(),
84 }
85 }
86}
87impl<'a> StreamsGroupHeartbeatRequest<'a> {
88 #[must_use]
89 pub fn to_owned(&self) -> streams_group_heartbeat_request::StreamsGroupHeartbeatRequest {
90 streams_group_heartbeat_request::StreamsGroupHeartbeatRequest {
91 group_id: (self.group_id).to_owned(),
92 member_id: (self.member_id).to_owned(),
93 member_epoch: (self.member_epoch),
94 endpoint_information_epoch: (self.endpoint_information_epoch),
95 instance_id: (self.instance_id).map(str::to_owned),
96 rack_id: (self.rack_id).map(str::to_owned),
97 rebalance_timeout_ms: (self.rebalance_timeout_ms),
98 topology: (self.topology).as_ref().map(Topology::to_owned),
99 active_tasks: (self.active_tasks).as_ref().map(|v| v.iter().map(super::common::streams_group_heartbeat_request::task_ids::TaskIds::to_owned).collect()),
100 standby_tasks: (self.standby_tasks).as_ref().map(|v| v.iter().map(super::common::streams_group_heartbeat_request::task_ids::TaskIds::to_owned).collect()),
101 warmup_tasks: (self.warmup_tasks).as_ref().map(|v| v.iter().map(super::common::streams_group_heartbeat_request::task_ids::TaskIds::to_owned).collect()),
102 process_id: (self.process_id).map(str::to_owned),
103 user_endpoint: (self.user_endpoint).as_ref().map(super::common::streams_group_heartbeat_request::endpoint::Endpoint::to_owned),
104 client_tags: (self.client_tags).as_ref().map(|v| v.iter().map(super::common::streams_group_heartbeat_request::key_value::KeyValue::to_owned).collect()),
105 task_offsets: (self.task_offsets).as_ref().map(|v| v.iter().map(super::common::streams_group_heartbeat_request::task_offset::TaskOffset::to_owned).collect()),
106 task_end_offsets: (self.task_end_offsets)
107 .as_ref()
108 .map(|v| v.iter().map(super::common::streams_group_heartbeat_request::task_offset::TaskOffset::to_owned).collect()),
109 shutdown_application: (self.shutdown_application),
110 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
111 }
112 }
113 fn encode_field_0<B: BufMut>(
114 &self,
115 buf: &mut B,
116 version: i16,
117 flex: bool,
118 ) -> Result<(), ProtocolError> {
119 if version >= 0 {
120 if flex {
121 put_compact_string(buf, self.group_id)?;
122 } else {
123 put_string(buf, self.group_id)?;
124 }
125 }
126 Ok(())
127 }
128 fn encode_field_1<B: BufMut>(
129 &self,
130 buf: &mut B,
131 version: i16,
132 flex: bool,
133 ) -> Result<(), ProtocolError> {
134 if version >= 0 {
135 if flex {
136 put_compact_string(buf, self.member_id)?;
137 } else {
138 put_string(buf, self.member_id)?;
139 }
140 }
141 Ok(())
142 }
143 fn encode_field_2<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
144 if version >= 0 {
145 put_i32(buf, self.member_epoch);
146 }
147 }
148 fn encode_field_3<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
149 if version >= 0 {
150 put_i32(buf, self.endpoint_information_epoch);
151 }
152 }
153 fn encode_field_4<B: BufMut>(
154 &self,
155 buf: &mut B,
156 version: i16,
157 flex: bool,
158 ) -> Result<(), ProtocolError> {
159 if version >= 0 {
160 if flex {
161 put_compact_nullable_string(buf, self.instance_id)?;
162 } else {
163 put_nullable_string(buf, self.instance_id)?;
164 }
165 }
166 Ok(())
167 }
168 fn encode_field_5<B: BufMut>(
169 &self,
170 buf: &mut B,
171 version: i16,
172 flex: bool,
173 ) -> Result<(), ProtocolError> {
174 if version >= 0 {
175 if flex {
176 put_compact_nullable_string(buf, self.rack_id)?;
177 } else {
178 put_nullable_string(buf, self.rack_id)?;
179 }
180 }
181 Ok(())
182 }
183 fn encode_field_6<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
184 if version >= 0 {
185 put_i32(buf, self.rebalance_timeout_ms);
186 }
187 }
188 fn encode_field_7<B: BufMut>(
189 &self,
190 buf: &mut B,
191 version: i16,
192 _flex: bool,
193 ) -> Result<(), ProtocolError> {
194 if version >= 0 {
195 match &self.topology {
196 None => {
197 buf.put_i8(-1);
198 }
199 Some(v) => {
200 buf.put_i8(1);
201 v.encode(buf, version)?;
202 }
203 }
204 }
205 Ok(())
206 }
207 fn encode_field_8<B: BufMut>(
208 &self,
209 buf: &mut B,
210 version: i16,
211 flex: bool,
212 ) -> Result<(), ProtocolError> {
213 if version >= 0 {
214 {
215 let len = (self.active_tasks).as_ref().map(Vec::len);
216 put_nullable_array_len(buf, len, flex)?;
217 if let Some(v) = &self.active_tasks {
218 for it in v {
219 it.encode(buf, version)?;
220 }
221 }
222 }
223 }
224 Ok(())
225 }
226 fn encode_field_9<B: BufMut>(
227 &self,
228 buf: &mut B,
229 version: i16,
230 flex: bool,
231 ) -> Result<(), ProtocolError> {
232 if version >= 0 {
233 {
234 let len = (self.standby_tasks).as_ref().map(Vec::len);
235 put_nullable_array_len(buf, len, flex)?;
236 if let Some(v) = &self.standby_tasks {
237 for it in v {
238 it.encode(buf, version)?;
239 }
240 }
241 }
242 }
243 Ok(())
244 }
245 fn encode_field_10<B: BufMut>(
246 &self,
247 buf: &mut B,
248 version: i16,
249 flex: bool,
250 ) -> Result<(), ProtocolError> {
251 if version >= 0 {
252 {
253 let len = (self.warmup_tasks).as_ref().map(Vec::len);
254 put_nullable_array_len(buf, len, flex)?;
255 if let Some(v) = &self.warmup_tasks {
256 for it in v {
257 it.encode(buf, version)?;
258 }
259 }
260 }
261 }
262 Ok(())
263 }
264 fn encode_field_11<B: BufMut>(
265 &self,
266 buf: &mut B,
267 version: i16,
268 flex: bool,
269 ) -> Result<(), ProtocolError> {
270 if version >= 0 {
271 if flex {
272 put_compact_nullable_string(buf, self.process_id)?;
273 } else {
274 put_nullable_string(buf, self.process_id)?;
275 }
276 }
277 Ok(())
278 }
279 fn encode_field_12<B: BufMut>(
280 &self,
281 buf: &mut B,
282 version: i16,
283 _flex: bool,
284 ) -> Result<(), ProtocolError> {
285 if version >= 0 {
286 match &self.user_endpoint {
287 None => {
288 buf.put_i8(-1);
289 }
290 Some(v) => {
291 buf.put_i8(1);
292 v.encode(buf, version)?;
293 }
294 }
295 }
296 Ok(())
297 }
298 fn encode_field_13<B: BufMut>(
299 &self,
300 buf: &mut B,
301 version: i16,
302 flex: bool,
303 ) -> Result<(), ProtocolError> {
304 if version >= 0 {
305 {
306 let len = (self.client_tags).as_ref().map(Vec::len);
307 put_nullable_array_len(buf, len, flex)?;
308 if let Some(v) = &self.client_tags {
309 for it in v {
310 it.encode(buf, version)?;
311 }
312 }
313 }
314 }
315 Ok(())
316 }
317 fn encode_field_14<B: BufMut>(
318 &self,
319 buf: &mut B,
320 version: i16,
321 flex: bool,
322 ) -> Result<(), ProtocolError> {
323 if version >= 0 {
324 {
325 let len = (self.task_offsets).as_ref().map(Vec::len);
326 put_nullable_array_len(buf, len, flex)?;
327 if let Some(v) = &self.task_offsets {
328 for it in v {
329 it.encode(buf, version)?;
330 }
331 }
332 }
333 }
334 Ok(())
335 }
336 fn encode_field_15<B: BufMut>(
337 &self,
338 buf: &mut B,
339 version: i16,
340 flex: bool,
341 ) -> Result<(), ProtocolError> {
342 if version >= 0 {
343 {
344 let len = (self.task_end_offsets).as_ref().map(Vec::len);
345 put_nullable_array_len(buf, len, flex)?;
346 if let Some(v) = &self.task_end_offsets {
347 for it in v {
348 it.encode(buf, version)?;
349 }
350 }
351 }
352 }
353 Ok(())
354 }
355 fn encode_field_16<B: BufMut>(&self, buf: &mut B, version: i16, _flex: bool) {
356 if version >= 0 {
357 put_bool(buf, self.shutdown_application);
358 }
359 }
360 fn encode_tagged_fields<B: BufMut>(
361 &self,
362 buf: &mut B,
363 _version: i16,
364 flex: bool,
365 ) -> Result<(), ProtocolError> {
366 if flex {
367 let tagged = WriteTaggedFields::new();
368 tagged.write(buf, &self.unknown_tagged_fields)?;
369 }
370 Ok(())
371 }
372 fn decode_field_0(
373 out: &mut Self,
374 buf: &mut &'a [u8],
375 version: i16,
376 flex: bool,
377 ) -> Result<(), ProtocolError> {
378 if version >= 0 {
379 out.group_id = if flex {
380 get_compact_string_borrowed(buf)?
381 } else {
382 get_string_borrowed(buf)?
383 };
384 }
385 Ok(())
386 }
387 fn decode_field_1(
388 out: &mut Self,
389 buf: &mut &'a [u8],
390 version: i16,
391 flex: bool,
392 ) -> Result<(), ProtocolError> {
393 if version >= 0 {
394 out.member_id = if flex {
395 get_compact_string_borrowed(buf)?
396 } else {
397 get_string_borrowed(buf)?
398 };
399 }
400 Ok(())
401 }
402 fn decode_field_2(
403 out: &mut Self,
404 buf: &mut &'a [u8],
405 version: i16,
406 _flex: bool,
407 ) -> Result<(), ProtocolError> {
408 if version >= 0 {
409 out.member_epoch = get_i32(buf)?;
410 }
411 Ok(())
412 }
413 fn decode_field_3(
414 out: &mut Self,
415 buf: &mut &'a [u8],
416 version: i16,
417 _flex: bool,
418 ) -> Result<(), ProtocolError> {
419 if version >= 0 {
420 out.endpoint_information_epoch = get_i32(buf)?;
421 }
422 Ok(())
423 }
424 fn decode_field_4(
425 out: &mut Self,
426 buf: &mut &'a [u8],
427 version: i16,
428 flex: bool,
429 ) -> Result<(), ProtocolError> {
430 if version >= 0 {
431 out.instance_id = if flex {
432 get_compact_nullable_string_borrowed(buf)?
433 } else {
434 get_nullable_string_borrowed(buf)?
435 };
436 }
437 Ok(())
438 }
439 fn decode_field_5(
440 out: &mut Self,
441 buf: &mut &'a [u8],
442 version: i16,
443 flex: bool,
444 ) -> Result<(), ProtocolError> {
445 if version >= 0 {
446 out.rack_id = if flex {
447 get_compact_nullable_string_borrowed(buf)?
448 } else {
449 get_nullable_string_borrowed(buf)?
450 };
451 }
452 Ok(())
453 }
454 fn decode_field_6(
455 out: &mut Self,
456 buf: &mut &'a [u8],
457 version: i16,
458 _flex: bool,
459 ) -> Result<(), ProtocolError> {
460 if version >= 0 {
461 out.rebalance_timeout_ms = get_i32(buf)?;
462 }
463 Ok(())
464 }
465 fn decode_field_7(
466 out: &mut Self,
467 buf: &mut &'a [u8],
468 version: i16,
469 _flex: bool,
470 ) -> Result<(), ProtocolError> {
471 if version >= 0 {
472 out.topology = if get_i8(buf)? < 0 {
473 None
474 } else {
475 Some(Topology::decode_borrow(buf, version)?)
476 };
477 }
478 Ok(())
479 }
480 fn decode_field_8(
481 out: &mut Self,
482 buf: &mut &'a [u8],
483 version: i16,
484 flex: bool,
485 ) -> Result<(), ProtocolError> {
486 if version >= 0 {
487 out.active_tasks = {
488 let opt = get_nullable_array_len(buf, flex)?;
489 match opt {
490 None => None,
491 Some(n) => {
492 let mut v = Vec::with_capacity(n);
493 for _ in 0..n {
494 v.push(super::common::streams_group_heartbeat_request::task_ids::TaskIds::decode_borrow(buf, version)?);
495 }
496 Some(v)
497 }
498 }
499 };
500 }
501 Ok(())
502 }
503 fn decode_field_9(
504 out: &mut Self,
505 buf: &mut &'a [u8],
506 version: i16,
507 flex: bool,
508 ) -> Result<(), ProtocolError> {
509 if version >= 0 {
510 out.standby_tasks = {
511 let opt = get_nullable_array_len(buf, flex)?;
512 match opt {
513 None => None,
514 Some(n) => {
515 let mut v = Vec::with_capacity(n);
516 for _ in 0..n {
517 v.push(super::common::streams_group_heartbeat_request::task_ids::TaskIds::decode_borrow(buf, version)?);
518 }
519 Some(v)
520 }
521 }
522 };
523 }
524 Ok(())
525 }
526 fn decode_field_10(
527 out: &mut Self,
528 buf: &mut &'a [u8],
529 version: i16,
530 flex: bool,
531 ) -> Result<(), ProtocolError> {
532 if version >= 0 {
533 out.warmup_tasks = {
534 let opt = get_nullable_array_len(buf, flex)?;
535 match opt {
536 None => None,
537 Some(n) => {
538 let mut v = Vec::with_capacity(n);
539 for _ in 0..n {
540 v.push(super::common::streams_group_heartbeat_request::task_ids::TaskIds::decode_borrow(buf, version)?);
541 }
542 Some(v)
543 }
544 }
545 };
546 }
547 Ok(())
548 }
549 fn decode_field_11(
550 out: &mut Self,
551 buf: &mut &'a [u8],
552 version: i16,
553 flex: bool,
554 ) -> Result<(), ProtocolError> {
555 if version >= 0 {
556 out.process_id = if flex {
557 get_compact_nullable_string_borrowed(buf)?
558 } else {
559 get_nullable_string_borrowed(buf)?
560 };
561 }
562 Ok(())
563 }
564 fn decode_field_12(
565 out: &mut Self,
566 buf: &mut &'a [u8],
567 version: i16,
568 _flex: bool,
569 ) -> Result<(), ProtocolError> {
570 if version >= 0 {
571 out.user_endpoint = if get_i8(buf)? < 0 {
572 None
573 } else {
574 Some(super::common::streams_group_heartbeat_request::endpoint::Endpoint::decode_borrow(buf, version)?)
575 };
576 }
577 Ok(())
578 }
579 fn decode_field_13(
580 out: &mut Self,
581 buf: &mut &'a [u8],
582 version: i16,
583 flex: bool,
584 ) -> Result<(), ProtocolError> {
585 if version >= 0 {
586 out.client_tags = {
587 let opt = get_nullable_array_len(buf, flex)?;
588 match opt {
589 None => None,
590 Some(n) => {
591 let mut v = Vec::with_capacity(n);
592 for _ in 0..n {
593 v.push(super::common::streams_group_heartbeat_request::key_value::KeyValue::decode_borrow(buf, version)?);
594 }
595 Some(v)
596 }
597 }
598 };
599 }
600 Ok(())
601 }
602 fn decode_field_14(
603 out: &mut Self,
604 buf: &mut &'a [u8],
605 version: i16,
606 flex: bool,
607 ) -> Result<(), ProtocolError> {
608 if version >= 0 {
609 out.task_offsets = {
610 let opt = get_nullable_array_len(buf, flex)?;
611 match opt {
612 None => None,
613 Some(n) => {
614 let mut v = Vec::with_capacity(n);
615 for _ in 0..n {
616 v.push(super::common::streams_group_heartbeat_request::task_offset::TaskOffset::decode_borrow(buf, version)?);
617 }
618 Some(v)
619 }
620 }
621 };
622 }
623 Ok(())
624 }
625 fn decode_field_15(
626 out: &mut Self,
627 buf: &mut &'a [u8],
628 version: i16,
629 flex: bool,
630 ) -> Result<(), ProtocolError> {
631 if version >= 0 {
632 out.task_end_offsets = {
633 let opt = get_nullable_array_len(buf, flex)?;
634 match opt {
635 None => None,
636 Some(n) => {
637 let mut v = Vec::with_capacity(n);
638 for _ in 0..n {
639 v.push(super::common::streams_group_heartbeat_request::task_offset::TaskOffset::decode_borrow(buf, version)?);
640 }
641 Some(v)
642 }
643 }
644 };
645 }
646 Ok(())
647 }
648 fn decode_field_16(
649 out: &mut Self,
650 buf: &mut &'a [u8],
651 version: i16,
652 _flex: bool,
653 ) -> Result<(), ProtocolError> {
654 if version >= 0 {
655 out.shutdown_application = get_bool(buf)?;
656 }
657 Ok(())
658 }
659 fn decode_tagged_fields(
660 out: &mut Self,
661 buf: &mut &'a [u8],
662 _version: i16,
663 flex: bool,
664 ) -> Result<(), ProtocolError> {
665 if flex {
666 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
667 }
668 Ok(())
669 }
670}
671impl Encode for StreamsGroupHeartbeatRequest<'_> {
672 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
673 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
674 return Err(ProtocolError::UnsupportedVersion {
675 api_key: API_KEY,
676 version,
677 });
678 }
679 let flex = is_flexible(version);
680 self.encode_field_0(buf, version, flex)?;
681 self.encode_field_1(buf, version, flex)?;
682 self.encode_field_2(buf, version, flex);
683 self.encode_field_3(buf, version, flex);
684 self.encode_field_4(buf, version, flex)?;
685 self.encode_field_5(buf, version, flex)?;
686 self.encode_field_6(buf, version, flex);
687 self.encode_field_7(buf, version, flex)?;
688 self.encode_field_8(buf, version, flex)?;
689 self.encode_field_9(buf, version, flex)?;
690 self.encode_field_10(buf, version, flex)?;
691 self.encode_field_11(buf, version, flex)?;
692 self.encode_field_12(buf, version, flex)?;
693 self.encode_field_13(buf, version, flex)?;
694 self.encode_field_14(buf, version, flex)?;
695 self.encode_field_15(buf, version, flex)?;
696 self.encode_field_16(buf, version, flex);
697 self.encode_tagged_fields(buf, version, flex)?;
698 Ok(())
699 }
700 fn encoded_len(&self, version: i16) -> usize {
701 let flex = is_flexible(version);
702 let mut n: usize = 0;
703 if version >= 0 {
704 n += if flex {
705 compact_string_len(self.group_id)
706 } else {
707 string_len(self.group_id)
708 };
709 }
710 if version >= 0 {
711 n += if flex {
712 compact_string_len(self.member_id)
713 } else {
714 string_len(self.member_id)
715 };
716 }
717 if version >= 0 {
718 n += 4;
719 }
720 if version >= 0 {
721 n += 4;
722 }
723 if version >= 0 {
724 n += if flex {
725 compact_nullable_string_len(self.instance_id)
726 } else {
727 nullable_string_len(self.instance_id)
728 };
729 }
730 if version >= 0 {
731 n += if flex {
732 compact_nullable_string_len(self.rack_id)
733 } else {
734 nullable_string_len(self.rack_id)
735 };
736 }
737 if version >= 0 {
738 n += 4;
739 }
740 if version >= 0 {
741 n += 1 + self.topology.as_ref().map_or(0, |v| v.encoded_len(version));
742 }
743 if version >= 0 {
744 n += {
745 let opt: Option<&Vec<_>> = (self.active_tasks).as_ref();
746 let prefix = nullable_array_len_prefix_len(opt.map(Vec::len), flex);
747 let body: usize =
748 opt.map_or(0, |v| v.iter().map(|it| it.encoded_len(version)).sum());
749 prefix + body
750 };
751 }
752 if version >= 0 {
753 n += {
754 let opt: Option<&Vec<_>> = (self.standby_tasks).as_ref();
755 let prefix = nullable_array_len_prefix_len(opt.map(Vec::len), flex);
756 let body: usize =
757 opt.map_or(0, |v| v.iter().map(|it| it.encoded_len(version)).sum());
758 prefix + body
759 };
760 }
761 if version >= 0 {
762 n += {
763 let opt: Option<&Vec<_>> = (self.warmup_tasks).as_ref();
764 let prefix = nullable_array_len_prefix_len(opt.map(Vec::len), flex);
765 let body: usize =
766 opt.map_or(0, |v| v.iter().map(|it| it.encoded_len(version)).sum());
767 prefix + body
768 };
769 }
770 if version >= 0 {
771 n += if flex {
772 compact_nullable_string_len(self.process_id)
773 } else {
774 nullable_string_len(self.process_id)
775 };
776 }
777 if version >= 0 {
778 n += 1 + self
779 .user_endpoint
780 .as_ref()
781 .map_or(0, |v| v.encoded_len(version));
782 }
783 if version >= 0 {
784 n += {
785 let opt: Option<&Vec<_>> = (self.client_tags).as_ref();
786 let prefix = nullable_array_len_prefix_len(opt.map(Vec::len), flex);
787 let body: usize =
788 opt.map_or(0, |v| v.iter().map(|it| it.encoded_len(version)).sum());
789 prefix + body
790 };
791 }
792 if version >= 0 {
793 n += {
794 let opt: Option<&Vec<_>> = (self.task_offsets).as_ref();
795 let prefix = nullable_array_len_prefix_len(opt.map(Vec::len), flex);
796 let body: usize =
797 opt.map_or(0, |v| v.iter().map(|it| it.encoded_len(version)).sum());
798 prefix + body
799 };
800 }
801 if version >= 0 {
802 n += {
803 let opt: Option<&Vec<_>> = (self.task_end_offsets).as_ref();
804 let prefix = nullable_array_len_prefix_len(opt.map(Vec::len), flex);
805 let body: usize =
806 opt.map_or(0, |v| v.iter().map(|it| it.encoded_len(version)).sum());
807 prefix + body
808 };
809 }
810 if version >= 0 {
811 n += 1;
812 }
813 if flex {
814 let known_pairs: Vec<(u32, usize)> = Vec::new();
815 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
816 }
817 n
818 }
819}
820impl<'de> DecodeBorrow<'de> for StreamsGroupHeartbeatRequest<'de> {
821 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
822 if !(MIN_VERSION..=MAX_VERSION).contains(&version) {
823 return Err(ProtocolError::UnsupportedVersion {
824 api_key: API_KEY,
825 version,
826 });
827 }
828 let flex = is_flexible(version);
829 let mut out = Self::default();
830 Self::decode_field_0(&mut out, buf, version, flex)?;
831 Self::decode_field_1(&mut out, buf, version, flex)?;
832 Self::decode_field_2(&mut out, buf, version, flex)?;
833 Self::decode_field_3(&mut out, buf, version, flex)?;
834 Self::decode_field_4(&mut out, buf, version, flex)?;
835 Self::decode_field_5(&mut out, buf, version, flex)?;
836 Self::decode_field_6(&mut out, buf, version, flex)?;
837 Self::decode_field_7(&mut out, buf, version, flex)?;
838 Self::decode_field_8(&mut out, buf, version, flex)?;
839 Self::decode_field_9(&mut out, buf, version, flex)?;
840 Self::decode_field_10(&mut out, buf, version, flex)?;
841 Self::decode_field_11(&mut out, buf, version, flex)?;
842 Self::decode_field_12(&mut out, buf, version, flex)?;
843 Self::decode_field_13(&mut out, buf, version, flex)?;
844 Self::decode_field_14(&mut out, buf, version, flex)?;
845 Self::decode_field_15(&mut out, buf, version, flex)?;
846 Self::decode_field_16(&mut out, buf, version, flex)?;
847 Self::decode_tagged_fields(&mut out, buf, version, flex)?;
848 Ok(out)
849 }
850}
851#[cfg(test)]
852impl StreamsGroupHeartbeatRequest<'_> {
853 #[must_use]
854 pub fn populated(version: i16) -> Self {
855 let mut m = Self::default();
856 if version >= 0 {
857 m.group_id = "x";
858 }
859 if version >= 0 {
860 m.member_id = "x";
861 }
862 if version >= 0 {
863 m.member_epoch = 1i32;
864 }
865 if version >= 0 {
866 m.endpoint_information_epoch = 1i32;
867 }
868 if version >= 0 {
869 m.instance_id = Some("x");
870 }
871 if version >= 0 {
872 m.rack_id = Some("x");
873 }
874 if version >= 0 {
875 m.rebalance_timeout_ms = 1i32;
876 }
877 if version >= 0 {
878 m.topology = Some(Topology::populated(version));
879 }
880 if version >= 0 {
881 m.active_tasks = Some(vec![
882 super::common::streams_group_heartbeat_request::task_ids::TaskIds::populated(
883 version,
884 ),
885 ]);
886 }
887 if version >= 0 {
888 m.standby_tasks = Some(vec![
889 super::common::streams_group_heartbeat_request::task_ids::TaskIds::populated(
890 version,
891 ),
892 ]);
893 }
894 if version >= 0 {
895 m.warmup_tasks = Some(vec![
896 super::common::streams_group_heartbeat_request::task_ids::TaskIds::populated(
897 version,
898 ),
899 ]);
900 }
901 if version >= 0 {
902 m.process_id = Some("x");
903 }
904 if version >= 0 {
905 m.user_endpoint = Some(
906 super::common::streams_group_heartbeat_request::endpoint::Endpoint::populated(
907 version,
908 ),
909 );
910 }
911 if version >= 0 {
912 m.client_tags = Some(vec![
913 super::common::streams_group_heartbeat_request::key_value::KeyValue::populated(
914 version,
915 ),
916 ]);
917 }
918 if version >= 0 {
919 m.task_offsets = Some(vec![
920 super::common::streams_group_heartbeat_request::task_offset::TaskOffset::populated(
921 version,
922 ),
923 ]);
924 }
925 if version >= 0 {
926 m.task_end_offsets = Some(vec![
927 super::common::streams_group_heartbeat_request::task_offset::TaskOffset::populated(
928 version,
929 ),
930 ]);
931 }
932 if version >= 0 {
933 m.shutdown_application = true;
934 }
935 m
936 }
937}
938#[derive(Debug, Clone, PartialEq, Eq, Default)]
939pub struct Topology<'a> {
940 pub epoch: i32,
941 pub subtopologies: Vec<Subtopology<'a>>,
942 pub unknown_tagged_fields: UnknownTaggedFields,
943}
944impl Topology<'_> {
945 #[must_use]
946 pub fn to_owned(&self) -> streams_group_heartbeat_request::Topology {
947 streams_group_heartbeat_request::Topology {
948 epoch: (self.epoch),
949 subtopologies: (self.subtopologies)
950 .iter()
951 .map(Subtopology::to_owned)
952 .collect(),
953 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
954 }
955 }
956}
957impl Encode for Topology<'_> {
958 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
959 let flex = version >= 0;
960 if version >= 0 {
961 put_i32(buf, self.epoch);
962 }
963 if version >= 0 {
964 {
965 put_array_len(buf, (self.subtopologies).len(), flex)?;
966 for it in &self.subtopologies {
967 it.encode(buf, version)?;
968 }
969 }
970 }
971 if flex {
972 let tagged = WriteTaggedFields::new();
973 tagged.write(buf, &self.unknown_tagged_fields)?;
974 }
975 Ok(())
976 }
977 fn encoded_len(&self, version: i16) -> usize {
978 let flex = version >= 0;
979 let mut n: usize = 0;
980 if version >= 0 {
981 n += 4;
982 }
983 if version >= 0 {
984 n += {
985 let prefix = array_len_prefix_len((self.subtopologies).len(), flex);
986 let body: usize = (self.subtopologies)
987 .iter()
988 .map(|it| it.encoded_len(version))
989 .sum();
990 prefix + body
991 };
992 }
993 if flex {
994 let known_pairs: Vec<(u32, usize)> = Vec::new();
995 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
996 }
997 n
998 }
999}
1000impl<'de> DecodeBorrow<'de> for Topology<'de> {
1001 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
1002 let flex = version >= 0;
1003 let mut out = Self::default();
1004 if version >= 0 {
1005 out.epoch = get_i32(buf)?;
1006 }
1007 if version >= 0 {
1008 out.subtopologies = {
1009 let n = get_array_len(buf, flex)?;
1010 let mut v = Vec::with_capacity(n);
1011 for _ in 0..n {
1012 v.push(Subtopology::decode_borrow(buf, version)?);
1013 }
1014 v
1015 };
1016 }
1017 if flex {
1018 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
1019 }
1020 Ok(out)
1021 }
1022}
1023#[cfg(test)]
1024impl Topology<'_> {
1025 #[must_use]
1026 pub fn populated(version: i16) -> Self {
1027 let mut m = Self::default();
1028 if version >= 0 {
1029 m.epoch = 1i32;
1030 }
1031 if version >= 0 {
1032 m.subtopologies = vec![Subtopology::populated(version)];
1033 }
1034 m
1035 }
1036}
1037#[derive(Debug, Clone, PartialEq, Eq, Default)]
1038pub struct Subtopology<'a> {
1039 pub subtopology_id: &'a str,
1040 pub source_topics: Vec<&'a str>,
1041 pub source_topic_regex: Vec<&'a str>,
1042 pub state_changelog_topics:
1043 Vec<super::common::streams_group_heartbeat_request::topic_info::TopicInfo<'a>>,
1044 pub repartition_sink_topics: Vec<&'a str>,
1045 pub repartition_source_topics:
1046 Vec<super::common::streams_group_heartbeat_request::topic_info::TopicInfo<'a>>,
1047 pub copartition_groups: Vec<CopartitionGroup>,
1048 pub unknown_tagged_fields: UnknownTaggedFields,
1049}
1050impl Subtopology<'_> {
1051 #[must_use]
1052 pub fn to_owned(&self) -> streams_group_heartbeat_request::Subtopology {
1053 streams_group_heartbeat_request::Subtopology {
1054 subtopology_id: (self.subtopology_id).to_owned(),
1055 source_topics: (self.source_topics).iter().map(|&s| s.to_owned()).collect(),
1056 source_topic_regex: (self.source_topic_regex)
1057 .iter()
1058 .map(|&s| s.to_owned())
1059 .collect(),
1060 state_changelog_topics: (self.state_changelog_topics)
1061 .iter()
1062 .map(
1063 super::common::streams_group_heartbeat_request::topic_info::TopicInfo::to_owned,
1064 )
1065 .collect(),
1066 repartition_sink_topics: (self.repartition_sink_topics)
1067 .iter()
1068 .map(|&s| s.to_owned())
1069 .collect(),
1070 repartition_source_topics: (self.repartition_source_topics)
1071 .iter()
1072 .map(
1073 super::common::streams_group_heartbeat_request::topic_info::TopicInfo::to_owned,
1074 )
1075 .collect(),
1076 copartition_groups: (self.copartition_groups)
1077 .iter()
1078 .map(CopartitionGroup::to_owned)
1079 .collect(),
1080 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
1081 }
1082 }
1083}
1084impl Encode for Subtopology<'_> {
1085 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
1086 let flex = version >= 0;
1087 if version >= 0 {
1088 if flex {
1089 put_compact_string(buf, self.subtopology_id)?;
1090 } else {
1091 put_string(buf, self.subtopology_id)?;
1092 }
1093 }
1094 if version >= 0 {
1095 {
1096 put_array_len(buf, (self.source_topics).len(), flex)?;
1097 for it in &self.source_topics {
1098 if flex {
1099 put_compact_string(buf, it)?;
1100 } else {
1101 put_string(buf, it)?;
1102 }
1103 }
1104 }
1105 }
1106 if version >= 0 {
1107 {
1108 put_array_len(buf, (self.source_topic_regex).len(), flex)?;
1109 for it in &self.source_topic_regex {
1110 if flex {
1111 put_compact_string(buf, it)?;
1112 } else {
1113 put_string(buf, it)?;
1114 }
1115 }
1116 }
1117 }
1118 if version >= 0 {
1119 {
1120 put_array_len(buf, (self.state_changelog_topics).len(), flex)?;
1121 for it in &self.state_changelog_topics {
1122 it.encode(buf, version)?;
1123 }
1124 }
1125 }
1126 if version >= 0 {
1127 {
1128 put_array_len(buf, (self.repartition_sink_topics).len(), flex)?;
1129 for it in &self.repartition_sink_topics {
1130 if flex {
1131 put_compact_string(buf, it)?;
1132 } else {
1133 put_string(buf, it)?;
1134 }
1135 }
1136 }
1137 }
1138 if version >= 0 {
1139 {
1140 put_array_len(buf, (self.repartition_source_topics).len(), flex)?;
1141 for it in &self.repartition_source_topics {
1142 it.encode(buf, version)?;
1143 }
1144 }
1145 }
1146 if version >= 0 {
1147 {
1148 put_array_len(buf, (self.copartition_groups).len(), flex)?;
1149 for it in &self.copartition_groups {
1150 it.encode(buf, version)?;
1151 }
1152 }
1153 }
1154 if flex {
1155 let tagged = WriteTaggedFields::new();
1156 tagged.write(buf, &self.unknown_tagged_fields)?;
1157 }
1158 Ok(())
1159 }
1160 fn encoded_len(&self, version: i16) -> usize {
1161 let flex = version >= 0;
1162 let mut n: usize = 0;
1163 if version >= 0 {
1164 n += if flex {
1165 compact_string_len(self.subtopology_id)
1166 } else {
1167 string_len(self.subtopology_id)
1168 };
1169 }
1170 if version >= 0 {
1171 n += {
1172 let prefix = array_len_prefix_len((self.source_topics).len(), flex);
1173 let body: usize = (self.source_topics)
1174 .iter()
1175 .map(|it| {
1176 if flex {
1177 compact_string_len(it)
1178 } else {
1179 string_len(it)
1180 }
1181 })
1182 .sum();
1183 prefix + body
1184 };
1185 }
1186 if version >= 0 {
1187 n += {
1188 let prefix = array_len_prefix_len((self.source_topic_regex).len(), flex);
1189 let body: usize = (self.source_topic_regex)
1190 .iter()
1191 .map(|it| {
1192 if flex {
1193 compact_string_len(it)
1194 } else {
1195 string_len(it)
1196 }
1197 })
1198 .sum();
1199 prefix + body
1200 };
1201 }
1202 if version >= 0 {
1203 n += {
1204 let prefix = array_len_prefix_len((self.state_changelog_topics).len(), flex);
1205 let body: usize = (self.state_changelog_topics)
1206 .iter()
1207 .map(|it| it.encoded_len(version))
1208 .sum();
1209 prefix + body
1210 };
1211 }
1212 if version >= 0 {
1213 n += {
1214 let prefix = array_len_prefix_len((self.repartition_sink_topics).len(), flex);
1215 let body: usize = (self.repartition_sink_topics)
1216 .iter()
1217 .map(|it| {
1218 if flex {
1219 compact_string_len(it)
1220 } else {
1221 string_len(it)
1222 }
1223 })
1224 .sum();
1225 prefix + body
1226 };
1227 }
1228 if version >= 0 {
1229 n += {
1230 let prefix = array_len_prefix_len((self.repartition_source_topics).len(), flex);
1231 let body: usize = (self.repartition_source_topics)
1232 .iter()
1233 .map(|it| it.encoded_len(version))
1234 .sum();
1235 prefix + body
1236 };
1237 }
1238 if version >= 0 {
1239 n += {
1240 let prefix = array_len_prefix_len((self.copartition_groups).len(), flex);
1241 let body: usize = (self.copartition_groups)
1242 .iter()
1243 .map(|it| it.encoded_len(version))
1244 .sum();
1245 prefix + body
1246 };
1247 }
1248 if flex {
1249 let known_pairs: Vec<(u32, usize)> = Vec::new();
1250 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
1251 }
1252 n
1253 }
1254}
1255impl<'de> DecodeBorrow<'de> for Subtopology<'de> {
1256 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
1257 let flex = version >= 0;
1258 let mut out = Self::default();
1259 if version >= 0 {
1260 out.subtopology_id = if flex {
1261 get_compact_string_borrowed(buf)?
1262 } else {
1263 get_string_borrowed(buf)?
1264 };
1265 }
1266 if version >= 0 {
1267 out.source_topics = {
1268 let n = get_array_len(buf, flex)?;
1269 let mut v = Vec::with_capacity(n);
1270 for _ in 0..n {
1271 v.push(if flex {
1272 get_compact_string_borrowed(buf)?
1273 } else {
1274 get_string_borrowed(buf)?
1275 });
1276 }
1277 v
1278 };
1279 }
1280 if version >= 0 {
1281 out.source_topic_regex = {
1282 let n = get_array_len(buf, flex)?;
1283 let mut v = Vec::with_capacity(n);
1284 for _ in 0..n {
1285 v.push(if flex {
1286 get_compact_string_borrowed(buf)?
1287 } else {
1288 get_string_borrowed(buf)?
1289 });
1290 }
1291 v
1292 };
1293 }
1294 if version >= 0 {
1295 out.state_changelog_topics = {
1296 let n = get_array_len(buf, flex)?;
1297 let mut v = Vec::with_capacity(n);
1298 for _ in 0..n {
1299 v.push(super::common::streams_group_heartbeat_request::topic_info::TopicInfo::decode_borrow(buf, version)?);
1300 }
1301 v
1302 };
1303 }
1304 if version >= 0 {
1305 out.repartition_sink_topics = {
1306 let n = get_array_len(buf, flex)?;
1307 let mut v = Vec::with_capacity(n);
1308 for _ in 0..n {
1309 v.push(if flex {
1310 get_compact_string_borrowed(buf)?
1311 } else {
1312 get_string_borrowed(buf)?
1313 });
1314 }
1315 v
1316 };
1317 }
1318 if version >= 0 {
1319 out.repartition_source_topics = {
1320 let n = get_array_len(buf, flex)?;
1321 let mut v = Vec::with_capacity(n);
1322 for _ in 0..n {
1323 v.push(super::common::streams_group_heartbeat_request::topic_info::TopicInfo::decode_borrow(buf, version)?);
1324 }
1325 v
1326 };
1327 }
1328 if version >= 0 {
1329 out.copartition_groups = {
1330 let n = get_array_len(buf, flex)?;
1331 let mut v = Vec::with_capacity(n);
1332 for _ in 0..n {
1333 v.push(CopartitionGroup::decode_borrow(buf, version)?);
1334 }
1335 v
1336 };
1337 }
1338 if flex {
1339 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
1340 }
1341 Ok(out)
1342 }
1343}
1344#[cfg(test)]
1345impl Subtopology<'_> {
1346 #[must_use]
1347 pub fn populated(version: i16) -> Self {
1348 let mut m = Self::default();
1349 if version >= 0 {
1350 m.subtopology_id = "x";
1351 }
1352 if version >= 0 {
1353 m.source_topics = vec!["x"];
1354 }
1355 if version >= 0 {
1356 m.source_topic_regex = vec!["x"];
1357 }
1358 if version >= 0 {
1359 m.state_changelog_topics = vec![
1360 super::common::streams_group_heartbeat_request::topic_info::TopicInfo::populated(
1361 version,
1362 ),
1363 ];
1364 }
1365 if version >= 0 {
1366 m.repartition_sink_topics = vec!["x"];
1367 }
1368 if version >= 0 {
1369 m.repartition_source_topics = vec![
1370 super::common::streams_group_heartbeat_request::topic_info::TopicInfo::populated(
1371 version,
1372 ),
1373 ];
1374 }
1375 if version >= 0 {
1376 m.copartition_groups = vec![CopartitionGroup::populated(version)];
1377 }
1378 m
1379 }
1380}
1381#[derive(Debug, Clone, PartialEq, Eq, Default)]
1382pub struct CopartitionGroup {
1383 pub source_topics: Vec<i16>,
1384 pub source_topic_regex: Vec<i16>,
1385 pub repartition_source_topics: Vec<i16>,
1386 pub unknown_tagged_fields: UnknownTaggedFields,
1387}
1388impl CopartitionGroup {
1389 #[must_use]
1390 pub fn to_owned(&self) -> streams_group_heartbeat_request::CopartitionGroup {
1391 streams_group_heartbeat_request::CopartitionGroup {
1392 source_topics: (self.source_topics).clone(),
1393 source_topic_regex: (self.source_topic_regex).clone(),
1394 repartition_source_topics: (self.repartition_source_topics).clone(),
1395 unknown_tagged_fields: self.unknown_tagged_fields.clone(),
1396 }
1397 }
1398}
1399impl Encode for CopartitionGroup {
1400 fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
1401 let flex = version >= 0;
1402 if version >= 0 {
1403 {
1404 put_array_len(buf, (self.source_topics).len(), flex)?;
1405 for it in &self.source_topics {
1406 put_i16(buf, *it);
1407 }
1408 }
1409 }
1410 if version >= 0 {
1411 {
1412 put_array_len(buf, (self.source_topic_regex).len(), flex)?;
1413 for it in &self.source_topic_regex {
1414 put_i16(buf, *it);
1415 }
1416 }
1417 }
1418 if version >= 0 {
1419 {
1420 put_array_len(buf, (self.repartition_source_topics).len(), flex)?;
1421 for it in &self.repartition_source_topics {
1422 put_i16(buf, *it);
1423 }
1424 }
1425 }
1426 if flex {
1427 let tagged = WriteTaggedFields::new();
1428 tagged.write(buf, &self.unknown_tagged_fields)?;
1429 }
1430 Ok(())
1431 }
1432 fn encoded_len(&self, version: i16) -> usize {
1433 let flex = version >= 0;
1434 let mut n: usize = 0;
1435 if version >= 0 {
1436 n += {
1437 let prefix = array_len_prefix_len((self.source_topics).len(), flex);
1438 let body: usize = (self.source_topics).iter().map(|_| 2).sum();
1439 prefix + body
1440 };
1441 }
1442 if version >= 0 {
1443 n += {
1444 let prefix = array_len_prefix_len((self.source_topic_regex).len(), flex);
1445 let body: usize = (self.source_topic_regex).iter().map(|_| 2).sum();
1446 prefix + body
1447 };
1448 }
1449 if version >= 0 {
1450 n += {
1451 let prefix = array_len_prefix_len((self.repartition_source_topics).len(), flex);
1452 let body: usize = (self.repartition_source_topics).iter().map(|_| 2).sum();
1453 prefix + body
1454 };
1455 }
1456 if flex {
1457 let known_pairs: Vec<(u32, usize)> = Vec::new();
1458 n += tagged_fields_len(&known_pairs, &self.unknown_tagged_fields);
1459 }
1460 n
1461 }
1462}
1463impl<'de> DecodeBorrow<'de> for CopartitionGroup {
1464 fn decode_borrow(buf: &mut &'de [u8], version: i16) -> Result<Self, ProtocolError> {
1465 let flex = version >= 0;
1466 let mut out = Self::default();
1467 if version >= 0 {
1468 out.source_topics = {
1469 let n = get_array_len(buf, flex)?;
1470 let mut v = Vec::with_capacity(n);
1471 for _ in 0..n {
1472 v.push(get_i16(buf)?);
1473 }
1474 v
1475 };
1476 }
1477 if version >= 0 {
1478 out.source_topic_regex = {
1479 let n = get_array_len(buf, flex)?;
1480 let mut v = Vec::with_capacity(n);
1481 for _ in 0..n {
1482 v.push(get_i16(buf)?);
1483 }
1484 v
1485 };
1486 }
1487 if version >= 0 {
1488 out.repartition_source_topics = {
1489 let n = get_array_len(buf, flex)?;
1490 let mut v = Vec::with_capacity(n);
1491 for _ in 0..n {
1492 v.push(get_i16(buf)?);
1493 }
1494 v
1495 };
1496 }
1497 if flex {
1498 out.unknown_tagged_fields = read_tagged_fields(buf, |_tag, _payload| Ok(false))?;
1499 }
1500 Ok(out)
1501 }
1502}
1503#[cfg(test)]
1504impl CopartitionGroup {
1505 #[must_use]
1506 pub fn populated(version: i16) -> Self {
1507 let mut m = Self::default();
1508 if version >= 0 {
1509 m.source_topics = vec![1i16];
1510 }
1511 if version >= 0 {
1512 m.source_topic_regex = vec![1i16];
1513 }
1514 if version >= 0 {
1515 m.repartition_source_topics = vec![1i16];
1516 }
1517 m
1518 }
1519}