sarama/consumer_group_describe_request.go at main · IBM/sarama · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
package sarama

// ConsumerGroupDescribeRequest describes consumer groups using the KIP-848 protocol.
type ConsumerGroupDescribeRequest struct {
	Version                     int16
	GroupIDs                    []string
	IncludeAuthorizedOperations bool
}

func NewConsumerGroupDescribeRequest(version KafkaVersion) *ConsumerGroupDescribeRequest {
	req := &ConsumerGroupDescribeRequest{}

	switch {
	case version.IsAtLeast(V4_0_0_0):
		req.Version = 1
	default:
		req.Version = 0
	}

	return req
}

func (r *ConsumerGroupDescribeRequest) encode(pe packetEncoder) error {
	if !r.isValidVersion() {
		return PacketEncodingError{"invalid or unsupported ConsumerGroupDescribeRequest version"}
	}

	if err := pe.putStringArray(r.GroupIDs); err != nil {
		return err
	}

	pe.putBool(r.IncludeAuthorizedOperations)

	pe.putEmptyTaggedFieldArray()

	return nil
}

func (r *ConsumerGroupDescribeRequest) decode(pd packetDecoder, version int16) (err error) {
	r.Version = version

	if !r.isValidVersion() {
		return PacketDecodingError{"invalid or unsupported ConsumerGroupDescribeRequest version"}
	}

	if r.GroupIDs, err = pd.getStringArray(); err != nil {
		return err
	}

	if r.IncludeAuthorizedOperations, err = pd.getBool(); err != nil {
		return err
	}

	_, err = pd.getEmptyTaggedFieldArray()

	return err
}

func (r *ConsumerGroupDescribeRequest) key() int16 { return apiKeyConsumerGroupDescribe }

func (r *ConsumerGroupDescribeRequest) version() int16 { return r.Version }

func (r *ConsumerGroupDescribeRequest) setVersion(v int16) { r.Version = v }

func (r *ConsumerGroupDescribeRequest) headerVersion() int16 { return 2 }

func (r *ConsumerGroupDescribeRequest) isValidVersion() bool { return r.Version >= 0 && r.Version <= 1 }

func (r *ConsumerGroupDescribeRequest) isFlexible() bool { return true }

func (r *ConsumerGroupDescribeRequest) isFlexibleVersion(version int16) bool { return version >= 0 }

func (r *ConsumerGroupDescribeRequest) requiredVersion() KafkaVersion {
	switch r.Version {
	case 1:
		return V4_0_0_0
	default:
		return V3_7_0_0
	}
}