forked from segmentio/kafka-go
-
Notifications
You must be signed in to change notification settings - Fork 0
/
describeconfigs.go
129 lines (106 loc) · 3.61 KB
/
describeconfigs.go
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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
package describeconfigs
import (
"strconv"
"github.com/segmentio/kafka-go/protocol"
)
const (
resourceTypeBroker int8 = 4
)
func init() {
protocol.Register(&Request{}, &Response{})
}
// Detailed API definition: https://kafka.apache.org/protocol#The_Messages_DescribeConfigs
type Request struct {
Resources []RequestResource `kafka:"min=v0,max=v3"`
IncludeSynonyms bool `kafka:"min=v1,max=v3"`
IncludeDocumentation bool `kafka:"min=v3,max=v3"`
}
func (r *Request) ApiKey() protocol.ApiKey { return protocol.DescribeConfigs }
func (r *Request) Broker(cluster protocol.Cluster) (protocol.Broker, error) {
// Broker metadata requests must be sent to the associated broker
for _, resource := range r.Resources {
if resource.ResourceType == resourceTypeBroker {
brokerID, err := strconv.Atoi(resource.ResourceName)
if err != nil {
return protocol.Broker{}, err
}
return cluster.Brokers[int32(brokerID)], nil
}
}
return cluster.Brokers[cluster.Controller], nil
}
func (r *Request) Split(cluster protocol.Cluster) (
[]protocol.Message,
protocol.Merger,
error,
) {
messages := []protocol.Message{}
topicsMessage := Request{}
for _, resource := range r.Resources {
// Split out broker requests to separate brokers
if resource.ResourceType == resourceTypeBroker {
messages = append(messages, &Request{
Resources: []RequestResource{resource},
})
} else {
topicsMessage.Resources = append(
topicsMessage.Resources, resource,
)
}
}
if len(topicsMessage.Resources) > 0 {
messages = append(messages, &topicsMessage)
}
return messages, new(Response), nil
}
type RequestResource struct {
ResourceType int8 `kafka:"min=v0,max=v3"`
ResourceName string `kafka:"min=v0,max=v3"`
ConfigNames []string `kafka:"min=v0,max=v3,nullable"`
}
type Response struct {
ThrottleTimeMs int32 `kafka:"min=v0,max=v3"`
Resources []ResponseResource `kafka:"min=v0,max=v3"`
}
func (r *Response) ApiKey() protocol.ApiKey { return protocol.DescribeConfigs }
func (r *Response) Merge(requests []protocol.Message, results []interface{}) (
protocol.Message,
error,
) {
response := &Response{}
for _, result := range results {
m, err := protocol.Result(result)
if err != nil {
return nil, err
}
response.Resources = append(
response.Resources,
m.(*Response).Resources...,
)
}
return response, nil
}
type ResponseResource struct {
ErrorCode int16 `kafka:"min=v0,max=v3"`
ErrorMessage string `kafka:"min=v0,max=v3,nullable"`
ResourceType int8 `kafka:"min=v0,max=v3"`
ResourceName string `kafka:"min=v0,max=v3"`
ConfigEntries []ResponseConfigEntry `kafka:"min=v0,max=v3"`
}
type ResponseConfigEntry struct {
ConfigName string `kafka:"min=v0,max=v3"`
ConfigValue string `kafka:"min=v0,max=v3,nullable"`
ReadOnly bool `kafka:"min=v0,max=v3"`
IsDefault bool `kafka:"min=v0,max=v0"`
ConfigSource int8 `kafka:"min=v1,max=v3"`
IsSensitive bool `kafka:"min=v0,max=v3"`
ConfigSynonyms []ResponseConfigSynonym `kafka:"min=v1,max=v3"`
ConfigType int8 `kafka:"min=v3,max=v3"`
ConfigDocumentation string `kafka:"min=v3,max=v3,nullable"`
}
type ResponseConfigSynonym struct {
ConfigName string `kafka:"min=v1,max=v3"`
ConfigValue string `kafka:"min=v1,max=v3,nullable"`
ConfigSource int8 `kafka:"min=v1,max=v3"`
}
var _ protocol.BrokerMessage = (*Request)(nil)