-
-
Notifications
You must be signed in to change notification settings - Fork 174
/
interface.go
386 lines (351 loc) · 12.7 KB
/
interface.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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
// Package kmsg contains Kafka request and response types and autogenerated
// serialization and deserialization functions.
//
// This package reserves the right to add new fields to struct types as Kafka
// adds new fields over time without bumping the major version. New requests
// will also be added without bumping the major version. The major version will
// also NOT BE BUMPED if a field's type is changed. The major version of this
// package is only bumped if it is required to be bumped per the kgo package.
//
// Kafka has only once in its history changed a non-array field's type,
// changing a string to a pointer to a string. These types of changes are
// expected to be very uncommon, and this package is provided with the
// understanding that it is advanced and may require some very minor
// maintenance if a field's type changes.
//
// If you are using this package directly with kgo, you should either ALWAYS
// use New functions (or Default functions after creating structs, you should
// pin the max supported version. If you use New functions, you should have
// safe defaults as new fields are added. If you pin versions, you will avoid
// new fields being used. If you do neither of these, you may opt in to new
// fields that do not have safe zero value defaults, and this may lead to
// errors or unexpected results.
//
// All "Default" functions set non-Go-default field defaults. They do not set
// any fields whose default value is a Go default. Thus, Default functions will
// set -1, but not 0 nor nil. All "New" functions also set non-Go-default
// fields. Requests and Responses also have a "NewPtr" function that is the
// same as "New," but returns a pointer to the type.
//
// Most of this package is generated, but a few things are manual. What is
// manual: all interfaces, the RequestFormatter, record / message / record
// batch reading, and sticky member metadata serialization.
package kmsg
import (
"context"
"encoding/binary"
"errors"
"hash/crc32"
"github.com/twmb/franz-go/pkg/kbin"
)
// Requestor issues requests. Notably, the kgo.Client and kgo.Broker implements
// Requestor. All Requests in this package have a RequestWith function to have
// type-safe requests.
type Requestor interface {
// Request issues a Request and returns either a Response or an error.
Request(context.Context, Request) (Response, error)
}
// Request represents a type that can be requested to Kafka.
type Request interface {
// Key returns the protocol key for this message kind.
Key() int16
// MaxVersion returns the maximum protocol version this message
// supports.
//
// This function allows one to implement a client that chooses message
// versions based off of the max of a message's max version in the
// client and the broker's max supported version.
MaxVersion() int16
// SetVersion sets the version to use for this request and response.
SetVersion(int16)
// GetVersion returns the version currently set to use for the request
// and response.
GetVersion() int16
// IsFlexible returns whether the request at its current version is
// "flexible" as per the KIP-482.
IsFlexible() bool
// AppendTo appends this message in wire protocol form to a slice and
// returns the slice.
AppendTo([]byte) []byte
// ReadFrom parses all of the input slice into the response type.
//
// This should return an error if too little data is input.
ReadFrom([]byte) error
// ResponseKind returns an empty Response that is expected for
// this message request.
ResponseKind() Response
}
// AdminRequest represents a request that must be issued to Kafka controllers.
type AdminRequest interface {
// IsAdminRequest is a method attached to requests that must be
// issed to Kafka controllers.
IsAdminRequest()
Request
}
// GroupCoordinatorRequest represents a request that must be issued to a
// group coordinator.
type GroupCoordinatorRequest interface {
// IsGroupCoordinatorRequest is a method attached to requests that
// must be issued to group coordinators.
IsGroupCoordinatorRequest()
Request
}
// TxnCoordinatorRequest represents a request that must be issued to a
// transaction coordinator.
type TxnCoordinatorRequest interface {
// IsTxnCoordinatorRequest is a method attached to requests that
// must be issued to transaction coordinators.
IsTxnCoordinatorRequest()
Request
}
// Response represents a type that Kafka responds with.
type Response interface {
// Key returns the protocol key for this message kind.
Key() int16
// MaxVersion returns the maximum protocol version this message
// supports.
MaxVersion() int16
// SetVersion sets the version to use for this request and response.
SetVersion(int16)
// GetVersion returns the version currently set to use for the request
// and response.
GetVersion() int16
// IsFlexible returns whether the request at its current version is
// "flexible" as per the KIP-482.
IsFlexible() bool
// AppendTo appends this message in wire protocol form to a slice and
// returns the slice.
AppendTo([]byte) []byte
// ReadFrom parses all of the input slice into the response type.
//
// This should return an error if too little data is input.
ReadFrom([]byte) error
// RequestKind returns an empty Request that is expected for
// this message request.
RequestKind() Request
}
// ThrottleResponse represents a response that could have a throttle applied by
// Kafka.
//
// Kafka 2.0.0 switched throttles from being applied before responses to being
// applied after responses.
type ThrottleResponse interface {
// Throttle returns the response's throttle millis value and whether
// Kafka applies the throttle after the response.
Throttle() (int32, bool)
}
// RequestFormatter formats requests.
//
// The default empty struct works correctly, but can be extended with the
// NewRequestFormatter function.
type RequestFormatter struct {
clientID *string
}
// RequestFormatterOpt applys options to a RequestFormatter.
type RequestFormatterOpt interface {
apply(*RequestFormatter)
}
type formatterOpt struct{ fn func(*RequestFormatter) }
func (opt formatterOpt) apply(f *RequestFormatter) { opt.fn(f) }
// FormatterClientID attaches the given client ID to any issued request,
// minus controlled shutdown v0, which uses its own special format.
func FormatterClientID(id string) RequestFormatterOpt {
return formatterOpt{func(f *RequestFormatter) { f.clientID = &id }}
}
// NewRequestFormatter returns a RequestFormatter with the opts applied.
func NewRequestFormatter(opts ...RequestFormatterOpt) *RequestFormatter {
a := new(RequestFormatter)
for _, opt := range opts {
opt.apply(a)
}
return a
}
// AppendRequest appends a full message request to dst, returning the updated
// slice. This message is the full body that needs to be written to issue a
// Kafka request.
func (f *RequestFormatter) AppendRequest(
dst []byte,
r Request,
correlationID int32,
) []byte {
dst = append(dst, 0, 0, 0, 0) // reserve length
k := r.Key()
v := r.GetVersion()
dst = kbin.AppendInt16(dst, k)
dst = kbin.AppendInt16(dst, v)
dst = kbin.AppendInt32(dst, correlationID)
if k == 7 && v == 0 {
return dst
}
// Even with flexible versions, we do not use a compact client id.
// Clients issue ApiVersions immediately before knowing the broker
// version, and old brokers will not be able to understand a compact
// client id.
dst = kbin.AppendNullableString(dst, f.clientID)
// The flexible tags end the request header, and then begins the
// request body.
if r.IsFlexible() {
var numTags uint8
dst = append(dst, numTags)
if numTags != 0 {
}
}
// Now the request body.
dst = r.AppendTo(dst)
kbin.AppendInt32(dst[:0], int32(len(dst[4:])))
return dst
}
// StringPtr is a helper to return a pointer to a string.
func StringPtr(in string) *string {
return &in
}
// ReadRecords reads n records from in and returns them, returning
// kerr.ErrNotEnoughData if in does not contain enough data.
func ReadRecords(n int, in []byte) ([]Record, error) {
rs := make([]Record, n)
for i := 0; i < n; i++ {
length, used := kbin.Varint(in)
total := used + int(length)
if used == 0 || length < 0 || len(in) < total {
return nil, kbin.ErrNotEnoughData
}
if err := (&rs[i]).ReadFrom(in[:total]); err != nil {
return nil, err
}
in = in[total:]
}
return rs, nil
}
// ErrEncodedCRCMismatch is returned from reading record batches or message sets when
// any batch or set has an encoded crc that does not match a calculated crc.
var ErrEncodedCRCMismatch = errors.New("encoded crc does not match calculated crc")
// ErrEncodedLengthMismatch is returned from reading record batches or message
// sets when any batch or set has an encoded length that does not match the
// earlier read length of the batch / set.
var ErrEncodedLengthMismatch = errors.New("encoded length does not match read length")
var crc32c = crc32.MakeTable(crc32.Castagnoli) // record crc's use Castagnoli table
// ReadRecordBatches reads as many record batches as possible from in,
// discarding any final trailing record batch. This is intended to be used
// for processing RecordBatches from a FetchResponse, where Kafka, as an
// internal optimization, may include a partial final RecordBatch.
func ReadRecordBatches(in []byte) ([]RecordBatch, error) {
var bs []RecordBatch
for len(in) > 12 {
length := int32(binary.BigEndian.Uint32(in[8:]))
length += 12
if len(in) < int(length) {
return bs, nil
}
var b RecordBatch
if err := b.ReadFrom(in[:length]); err != nil {
return bs, nil
}
if int32(len(in[12:length])) != b.Length {
return bs, ErrEncodedLengthMismatch
}
// If we did not error, the length was at _least_ 21.
if int32(crc32.Checksum(in[21:length], crc32c)) != b.CRC {
return bs, ErrEncodedCRCMismatch
}
bs = append(bs, b)
in = in[length:]
}
return bs, nil
}
// ReadV1Messages reads as many v1 message sets as possible from
// in, discarding any final trailing message set. This is intended to be used
// for processing v1 MessageSets from a FetchResponse, where Kafka, as an
// internal optimization, may include a partial final MessageSet.
func ReadV1Messages(in []byte) ([]MessageV1, error) {
var ms []MessageV1
for len(in) > 12 {
length := int32(binary.BigEndian.Uint32(in[8:]))
length += 12
if len(in) < int(length) {
return ms, nil
}
var m MessageV1
if err := m.ReadFrom(in[:length]); err != nil {
return ms, nil
}
if int32(len(in[12:length])) != m.MessageSize {
return ms, ErrEncodedLengthMismatch
}
if int32(crc32.ChecksumIEEE(in[16:length])) != m.CRC {
return ms, ErrEncodedCRCMismatch
}
ms = append(ms, m)
in = in[length:]
}
return ms, nil
}
// ReadV0Messages reads as many v0 message sets as possible from
// in, discarding any final trailing message set. This is intended to be used
// for processing v0 MessageSets from a FetchResponse, where Kafka, as an
// internal optimization, may include a partial final MessageSet.
func ReadV0Messages(in []byte) ([]MessageV0, error) {
var ms []MessageV0
for len(in) > 12 {
length := int32(binary.BigEndian.Uint32(in[8:]))
length += 12
if len(in) < int(length) {
return ms, nil
}
var m MessageV0
if err := m.ReadFrom(in[:length]); err != nil {
return ms, nil
}
if int32(len(in[12:length])) != m.MessageSize {
return ms, ErrEncodedLengthMismatch
}
if int32(crc32.ChecksumIEEE(in[16:length])) != m.CRC {
return ms, ErrEncodedCRCMismatch
}
ms = append(ms, m)
in = in[length:]
}
return ms, nil
}
// ReadFrom provides decoding various versions of sticky member metadata. A key
// point of this type is that it does not contain a version number inside it,
// but it is versioned: if decoding v1 fails, this falls back to v0.
func (s *StickyMemberMetadata) ReadFrom(src []byte) error {
b := kbin.Reader{Src: src}
for i := b.ArrayLen(); i > 0; i-- {
var assignment StickyMemberMetadataCurrentAssignment
assignment.Topic = b.String()
for i := b.ArrayLen(); i > 0; i-- {
assignment.Partitions = append(assignment.Partitions, b.Int32())
}
s.CurrentAssignment = append(s.CurrentAssignment, assignment)
}
if len(b.Src) > 0 {
s.Generation = b.Int32()
} else {
s.Generation = -1
}
return b.Complete()
}
// AppendTo provides appending various versions of sticky member metadata to dst.
// If generation is not -1 (default for v0), this appends as version 1.
func (s *StickyMemberMetadata) AppendTo(dst []byte) []byte {
dst = kbin.AppendArrayLen(dst, len(s.CurrentAssignment))
for _, assignment := range s.CurrentAssignment {
dst = kbin.AppendString(dst, assignment.Topic)
dst = kbin.AppendArrayLen(dst, len(assignment.Partitions))
for _, partition := range assignment.Partitions {
dst = kbin.AppendInt32(dst, partition)
}
}
if s.Generation != -1 {
dst = kbin.AppendInt32(dst, s.Generation)
}
return dst
}
// SkipTags skips tags in a reader.
func SkipTags(b *kbin.Reader) {
for num := b.Uvarint(); num > 0; num-- {
_, size := b.Uvarint(), b.Uvarint()
b.Span(int(size))
}
}