blob: ed20522936c1659aaf24769e7c87f2f4e3ca1254 [file] [log] [blame]
khenaidooac637102019-01-14 15:44:34 -05001package sarama
2
3import (
4 "compress/gzip"
5 "crypto/tls"
6 "fmt"
7 "io/ioutil"
8 "net"
9 "regexp"
10 "time"
11
12 "github.com/rcrowley/go-metrics"
13)
14
15const defaultClientID = "sarama"
16
17var validID = regexp.MustCompile(`\A[A-Za-z0-9._-]+\z`)
18
19// Config is used to pass multiple configuration options to Sarama's constructors.
20type Config struct {
21 // Admin is the namespace for ClusterAdmin properties used by the administrative Kafka client.
22 Admin struct {
23 // The maximum duration the administrative Kafka client will wait for ClusterAdmin operations,
24 // including topics, brokers, configurations and ACLs (defaults to 3 seconds).
25 Timeout time.Duration
26 }
27
28 // Net is the namespace for network-level properties used by the Broker, and
29 // shared by the Client/Producer/Consumer.
30 Net struct {
31 // How many outstanding requests a connection is allowed to have before
32 // sending on it blocks (default 5).
33 MaxOpenRequests int
34
35 // All three of the below configurations are similar to the
36 // `socket.timeout.ms` setting in JVM kafka. All of them default
37 // to 30 seconds.
38 DialTimeout time.Duration // How long to wait for the initial connection.
39 ReadTimeout time.Duration // How long to wait for a response.
40 WriteTimeout time.Duration // How long to wait for a transmit.
41
42 TLS struct {
43 // Whether or not to use TLS when connecting to the broker
44 // (defaults to false).
45 Enable bool
46 // The TLS configuration to use for secure connections if
47 // enabled (defaults to nil).
48 Config *tls.Config
49 }
50
51 // SASL based authentication with broker. While there are multiple SASL authentication methods
52 // the current implementation is limited to plaintext (SASL/PLAIN) authentication
53 SASL struct {
54 // Whether or not to use SASL authentication when connecting to the broker
55 // (defaults to false).
56 Enable bool
57 // Whether or not to send the Kafka SASL handshake first if enabled
58 // (defaults to true). You should only set this to false if you're using
59 // a non-Kafka SASL proxy.
60 Handshake bool
61 //username and password for SASL/PLAIN authentication
62 User string
63 Password string
64 }
65
66 // KeepAlive specifies the keep-alive period for an active network connection.
67 // If zero, keep-alives are disabled. (default is 0: disabled).
68 KeepAlive time.Duration
69
70 // LocalAddr is the local address to use when dialing an
71 // address. The address must be of a compatible type for the
72 // network being dialed.
73 // If nil, a local address is automatically chosen.
74 LocalAddr net.Addr
75 }
76
77 // Metadata is the namespace for metadata management properties used by the
78 // Client, and shared by the Producer/Consumer.
79 Metadata struct {
80 Retry struct {
81 // The total number of times to retry a metadata request when the
82 // cluster is in the middle of a leader election (default 3).
83 Max int
84 // How long to wait for leader election to occur before retrying
85 // (default 250ms). Similar to the JVM's `retry.backoff.ms`.
86 Backoff time.Duration
87 }
88 // How frequently to refresh the cluster metadata in the background.
89 // Defaults to 10 minutes. Set to 0 to disable. Similar to
90 // `topic.metadata.refresh.interval.ms` in the JVM version.
91 RefreshFrequency time.Duration
92
93 // Whether to maintain a full set of metadata for all topics, or just
94 // the minimal set that has been necessary so far. The full set is simpler
95 // and usually more convenient, but can take up a substantial amount of
96 // memory if you have many topics and partitions. Defaults to true.
97 Full bool
98 }
99
100 // Producer is the namespace for configuration related to producing messages,
101 // used by the Producer.
102 Producer struct {
103 // The maximum permitted size of a message (defaults to 1000000). Should be
104 // set equal to or smaller than the broker's `message.max.bytes`.
105 MaxMessageBytes int
106 // The level of acknowledgement reliability needed from the broker (defaults
107 // to WaitForLocal). Equivalent to the `request.required.acks` setting of the
108 // JVM producer.
109 RequiredAcks RequiredAcks
110 // The maximum duration the broker will wait the receipt of the number of
111 // RequiredAcks (defaults to 10 seconds). This is only relevant when
112 // RequiredAcks is set to WaitForAll or a number > 1. Only supports
113 // millisecond resolution, nanoseconds will be truncated. Equivalent to
114 // the JVM producer's `request.timeout.ms` setting.
115 Timeout time.Duration
116 // The type of compression to use on messages (defaults to no compression).
117 // Similar to `compression.codec` setting of the JVM producer.
118 Compression CompressionCodec
119 // The level of compression to use on messages. The meaning depends
120 // on the actual compression type used and defaults to default compression
121 // level for the codec.
122 CompressionLevel int
123 // Generates partitioners for choosing the partition to send messages to
124 // (defaults to hashing the message key). Similar to the `partitioner.class`
125 // setting for the JVM producer.
126 Partitioner PartitionerConstructor
127 // If enabled, the producer will ensure that exactly one copy of each message is
128 // written.
129 Idempotent bool
130
131 // Return specifies what channels will be populated. If they are set to true,
132 // you must read from the respective channels to prevent deadlock. If,
133 // however, this config is used to create a `SyncProducer`, both must be set
134 // to true and you shall not read from the channels since the producer does
135 // this internally.
136 Return struct {
137 // If enabled, successfully delivered messages will be returned on the
138 // Successes channel (default disabled).
139 Successes bool
140
141 // If enabled, messages that failed to deliver will be returned on the
142 // Errors channel, including error (default enabled).
143 Errors bool
144 }
145
146 // The following config options control how often messages are batched up and
147 // sent to the broker. By default, messages are sent as fast as possible, and
148 // all messages received while the current batch is in-flight are placed
149 // into the subsequent batch.
150 Flush struct {
151 // The best-effort number of bytes needed to trigger a flush. Use the
152 // global sarama.MaxRequestSize to set a hard upper limit.
153 Bytes int
154 // The best-effort number of messages needed to trigger a flush. Use
155 // `MaxMessages` to set a hard upper limit.
156 Messages int
157 // The best-effort frequency of flushes. Equivalent to
158 // `queue.buffering.max.ms` setting of JVM producer.
159 Frequency time.Duration
160 // The maximum number of messages the producer will send in a single
161 // broker request. Defaults to 0 for unlimited. Similar to
162 // `queue.buffering.max.messages` in the JVM producer.
163 MaxMessages int
164 }
165
166 Retry struct {
167 // The total number of times to retry sending a message (default 3).
168 // Similar to the `message.send.max.retries` setting of the JVM producer.
169 Max int
170 // How long to wait for the cluster to settle between retries
171 // (default 100ms). Similar to the `retry.backoff.ms` setting of the
172 // JVM producer.
173 Backoff time.Duration
174 }
175 }
176
177 // Consumer is the namespace for configuration related to consuming messages,
178 // used by the Consumer.
179 Consumer struct {
180
181 // Group is the namespace for configuring consumer group.
182 Group struct {
183 Session struct {
184 // The timeout used to detect consumer failures when using Kafka's group management facility.
185 // The consumer sends periodic heartbeats to indicate its liveness to the broker.
186 // If no heartbeats are received by the broker before the expiration of this session timeout,
187 // then the broker will remove this consumer from the group and initiate a rebalance.
188 // Note that the value must be in the allowable range as configured in the broker configuration
189 // by `group.min.session.timeout.ms` and `group.max.session.timeout.ms` (default 10s)
190 Timeout time.Duration
191 }
192 Heartbeat struct {
193 // The expected time between heartbeats to the consumer coordinator when using Kafka's group
194 // management facilities. Heartbeats are used to ensure that the consumer's session stays active and
195 // to facilitate rebalancing when new consumers join or leave the group.
196 // The value must be set lower than Consumer.Group.Session.Timeout, but typically should be set no
197 // higher than 1/3 of that value.
198 // It can be adjusted even lower to control the expected time for normal rebalances (default 3s)
199 Interval time.Duration
200 }
201 Rebalance struct {
202 // Strategy for allocating topic partitions to members (default BalanceStrategyRange)
203 Strategy BalanceStrategy
204 // The maximum allowed time for each worker to join the group once a rebalance has begun.
205 // This is basically a limit on the amount of time needed for all tasks to flush any pending
206 // data and commit offsets. If the timeout is exceeded, then the worker will be removed from
207 // the group, which will cause offset commit failures (default 60s).
208 Timeout time.Duration
209
210 Retry struct {
211 // When a new consumer joins a consumer group the set of consumers attempt to "rebalance"
212 // the load to assign partitions to each consumer. If the set of consumers changes while
213 // this assignment is taking place the rebalance will fail and retry. This setting controls
214 // the maximum number of attempts before giving up (default 4).
215 Max int
216 // Backoff time between retries during rebalance (default 2s)
217 Backoff time.Duration
218 }
219 }
220 Member struct {
221 // Custom metadata to include when joining the group. The user data for all joined members
222 // can be retrieved by sending a DescribeGroupRequest to the broker that is the
223 // coordinator for the group.
224 UserData []byte
225 }
226 }
227
228 Retry struct {
229 // How long to wait after a failing to read from a partition before
230 // trying again (default 2s).
231 Backoff time.Duration
232 }
233
234 // Fetch is the namespace for controlling how many bytes are retrieved by any
235 // given request.
236 Fetch struct {
237 // The minimum number of message bytes to fetch in a request - the broker
238 // will wait until at least this many are available. The default is 1,
239 // as 0 causes the consumer to spin when no messages are available.
240 // Equivalent to the JVM's `fetch.min.bytes`.
241 Min int32
242 // The default number of message bytes to fetch from the broker in each
243 // request (default 1MB). This should be larger than the majority of
244 // your messages, or else the consumer will spend a lot of time
245 // negotiating sizes and not actually consuming. Similar to the JVM's
246 // `fetch.message.max.bytes`.
247 Default int32
248 // The maximum number of message bytes to fetch from the broker in a
249 // single request. Messages larger than this will return
250 // ErrMessageTooLarge and will not be consumable, so you must be sure
251 // this is at least as large as your largest message. Defaults to 0
252 // (no limit). Similar to the JVM's `fetch.message.max.bytes`. The
253 // global `sarama.MaxResponseSize` still applies.
254 Max int32
255 }
256 // The maximum amount of time the broker will wait for Consumer.Fetch.Min
257 // bytes to become available before it returns fewer than that anyways. The
258 // default is 250ms, since 0 causes the consumer to spin when no events are
259 // available. 100-500ms is a reasonable range for most cases. Kafka only
260 // supports precision up to milliseconds; nanoseconds will be truncated.
261 // Equivalent to the JVM's `fetch.wait.max.ms`.
262 MaxWaitTime time.Duration
263
264 // The maximum amount of time the consumer expects a message takes to
265 // process for the user. If writing to the Messages channel takes longer
266 // than this, that partition will stop fetching more messages until it
267 // can proceed again.
268 // Note that, since the Messages channel is buffered, the actual grace time is
269 // (MaxProcessingTime * ChanneBufferSize). Defaults to 100ms.
270 // If a message is not written to the Messages channel between two ticks
271 // of the expiryTicker then a timeout is detected.
272 // Using a ticker instead of a timer to detect timeouts should typically
273 // result in many fewer calls to Timer functions which may result in a
274 // significant performance improvement if many messages are being sent
275 // and timeouts are infrequent.
276 // The disadvantage of using a ticker instead of a timer is that
277 // timeouts will be less accurate. That is, the effective timeout could
278 // be between `MaxProcessingTime` and `2 * MaxProcessingTime`. For
279 // example, if `MaxProcessingTime` is 100ms then a delay of 180ms
280 // between two messages being sent may not be recognized as a timeout.
281 MaxProcessingTime time.Duration
282
283 // Return specifies what channels will be populated. If they are set to true,
284 // you must read from them to prevent deadlock.
285 Return struct {
286 // If enabled, any errors that occurred while consuming are returned on
287 // the Errors channel (default disabled).
288 Errors bool
289 }
290
291 // Offsets specifies configuration for how and when to commit consumed
292 // offsets. This currently requires the manual use of an OffsetManager
293 // but will eventually be automated.
294 Offsets struct {
295 // How frequently to commit updated offsets. Defaults to 1s.
296 CommitInterval time.Duration
297
298 // The initial offset to use if no offset was previously committed.
299 // Should be OffsetNewest or OffsetOldest. Defaults to OffsetNewest.
300 Initial int64
301
302 // The retention duration for committed offsets. If zero, disabled
303 // (in which case the `offsets.retention.minutes` option on the
304 // broker will be used). Kafka only supports precision up to
305 // milliseconds; nanoseconds will be truncated. Requires Kafka
306 // broker version 0.9.0 or later.
307 // (default is 0: disabled).
308 Retention time.Duration
309
310 Retry struct {
311 // The total number of times to retry failing commit
312 // requests during OffsetManager shutdown (default 3).
313 Max int
314 }
315 }
316 }
317
318 // A user-provided string sent with every request to the brokers for logging,
319 // debugging, and auditing purposes. Defaults to "sarama", but you should
320 // probably set it to something specific to your application.
321 ClientID string
322 // The number of events to buffer in internal and external channels. This
323 // permits the producer and consumer to continue processing some messages
324 // in the background while user code is working, greatly improving throughput.
325 // Defaults to 256.
326 ChannelBufferSize int
327 // The version of Kafka that Sarama will assume it is running against.
328 // Defaults to the oldest supported stable version. Since Kafka provides
329 // backwards-compatibility, setting it to a version older than you have
330 // will not break anything, although it may prevent you from using the
331 // latest features. Setting it to a version greater than you are actually
332 // running may lead to random breakage.
333 Version KafkaVersion
334 // The registry to define metrics into.
335 // Defaults to a local registry.
336 // If you want to disable metrics gathering, set "metrics.UseNilMetrics" to "true"
337 // prior to starting Sarama.
338 // See Examples on how to use the metrics registry
339 MetricRegistry metrics.Registry
340}
341
342// NewConfig returns a new configuration instance with sane defaults.
343func NewConfig() *Config {
344 c := &Config{}
345
346 c.Admin.Timeout = 3 * time.Second
347
348 c.Net.MaxOpenRequests = 5
349 c.Net.DialTimeout = 30 * time.Second
350 c.Net.ReadTimeout = 30 * time.Second
351 c.Net.WriteTimeout = 30 * time.Second
352 c.Net.SASL.Handshake = true
353
354 c.Metadata.Retry.Max = 3
355 c.Metadata.Retry.Backoff = 250 * time.Millisecond
356 c.Metadata.RefreshFrequency = 10 * time.Minute
357 c.Metadata.Full = true
358
359 c.Producer.MaxMessageBytes = 1000000
360 c.Producer.RequiredAcks = WaitForLocal
361 c.Producer.Timeout = 10 * time.Second
362 c.Producer.Partitioner = NewHashPartitioner
363 c.Producer.Retry.Max = 3
364 c.Producer.Retry.Backoff = 100 * time.Millisecond
365 c.Producer.Return.Errors = true
366 c.Producer.CompressionLevel = CompressionLevelDefault
367
368 c.Consumer.Fetch.Min = 1
369 c.Consumer.Fetch.Default = 1024 * 1024
370 c.Consumer.Retry.Backoff = 2 * time.Second
371 c.Consumer.MaxWaitTime = 250 * time.Millisecond
372 c.Consumer.MaxProcessingTime = 100 * time.Millisecond
373 c.Consumer.Return.Errors = false
374 c.Consumer.Offsets.CommitInterval = 1 * time.Second
375 c.Consumer.Offsets.Initial = OffsetNewest
376 c.Consumer.Offsets.Retry.Max = 3
377
378 c.Consumer.Group.Session.Timeout = 10 * time.Second
379 c.Consumer.Group.Heartbeat.Interval = 3 * time.Second
380 c.Consumer.Group.Rebalance.Strategy = BalanceStrategyRange
381 c.Consumer.Group.Rebalance.Timeout = 60 * time.Second
382 c.Consumer.Group.Rebalance.Retry.Max = 4
383 c.Consumer.Group.Rebalance.Retry.Backoff = 2 * time.Second
384
385 c.ClientID = defaultClientID
386 c.ChannelBufferSize = 256
387 c.Version = MinVersion
388 c.MetricRegistry = metrics.NewRegistry()
389
390 return c
391}
392
393// Validate checks a Config instance. It will return a
394// ConfigurationError if the specified values don't make sense.
395func (c *Config) Validate() error {
396 // some configuration values should be warned on but not fail completely, do those first
397 if c.Net.TLS.Enable == false && c.Net.TLS.Config != nil {
398 Logger.Println("Net.TLS is disabled but a non-nil configuration was provided.")
399 }
400 if c.Net.SASL.Enable == false {
401 if c.Net.SASL.User != "" {
402 Logger.Println("Net.SASL is disabled but a non-empty username was provided.")
403 }
404 if c.Net.SASL.Password != "" {
405 Logger.Println("Net.SASL is disabled but a non-empty password was provided.")
406 }
407 }
408 if c.Producer.RequiredAcks > 1 {
409 Logger.Println("Producer.RequiredAcks > 1 is deprecated and will raise an exception with kafka >= 0.8.2.0.")
410 }
411 if c.Producer.MaxMessageBytes >= int(MaxRequestSize) {
412 Logger.Println("Producer.MaxMessageBytes must be smaller than MaxRequestSize; it will be ignored.")
413 }
414 if c.Producer.Flush.Bytes >= int(MaxRequestSize) {
415 Logger.Println("Producer.Flush.Bytes must be smaller than MaxRequestSize; it will be ignored.")
416 }
417 if (c.Producer.Flush.Bytes > 0 || c.Producer.Flush.Messages > 0) && c.Producer.Flush.Frequency == 0 {
418 Logger.Println("Producer.Flush: Bytes or Messages are set, but Frequency is not; messages may not get flushed.")
419 }
420 if c.Producer.Timeout%time.Millisecond != 0 {
421 Logger.Println("Producer.Timeout only supports millisecond resolution; nanoseconds will be truncated.")
422 }
423 if c.Consumer.MaxWaitTime < 100*time.Millisecond {
424 Logger.Println("Consumer.MaxWaitTime is very low, which can cause high CPU and network usage. See documentation for details.")
425 }
426 if c.Consumer.MaxWaitTime%time.Millisecond != 0 {
427 Logger.Println("Consumer.MaxWaitTime only supports millisecond precision; nanoseconds will be truncated.")
428 }
429 if c.Consumer.Offsets.Retention%time.Millisecond != 0 {
430 Logger.Println("Consumer.Offsets.Retention only supports millisecond precision; nanoseconds will be truncated.")
431 }
432 if c.Consumer.Group.Session.Timeout%time.Millisecond != 0 {
433 Logger.Println("Consumer.Group.Session.Timeout only supports millisecond precision; nanoseconds will be truncated.")
434 }
435 if c.Consumer.Group.Heartbeat.Interval%time.Millisecond != 0 {
436 Logger.Println("Consumer.Group.Heartbeat.Interval only supports millisecond precision; nanoseconds will be truncated.")
437 }
438 if c.Consumer.Group.Rebalance.Timeout%time.Millisecond != 0 {
439 Logger.Println("Consumer.Group.Rebalance.Timeout only supports millisecond precision; nanoseconds will be truncated.")
440 }
441 if c.ClientID == defaultClientID {
442 Logger.Println("ClientID is the default of 'sarama', you should consider setting it to something application-specific.")
443 }
444
445 // validate Net values
446 switch {
447 case c.Net.MaxOpenRequests <= 0:
448 return ConfigurationError("Net.MaxOpenRequests must be > 0")
449 case c.Net.DialTimeout <= 0:
450 return ConfigurationError("Net.DialTimeout must be > 0")
451 case c.Net.ReadTimeout <= 0:
452 return ConfigurationError("Net.ReadTimeout must be > 0")
453 case c.Net.WriteTimeout <= 0:
454 return ConfigurationError("Net.WriteTimeout must be > 0")
455 case c.Net.KeepAlive < 0:
456 return ConfigurationError("Net.KeepAlive must be >= 0")
457 case c.Net.SASL.Enable == true && c.Net.SASL.User == "":
458 return ConfigurationError("Net.SASL.User must not be empty when SASL is enabled")
459 case c.Net.SASL.Enable == true && c.Net.SASL.Password == "":
460 return ConfigurationError("Net.SASL.Password must not be empty when SASL is enabled")
461 }
462
463 // validate the Admin values
464 switch {
465 case c.Admin.Timeout <= 0:
466 return ConfigurationError("Admin.Timeout must be > 0")
467 }
468
469 // validate the Metadata values
470 switch {
471 case c.Metadata.Retry.Max < 0:
472 return ConfigurationError("Metadata.Retry.Max must be >= 0")
473 case c.Metadata.Retry.Backoff < 0:
474 return ConfigurationError("Metadata.Retry.Backoff must be >= 0")
475 case c.Metadata.RefreshFrequency < 0:
476 return ConfigurationError("Metadata.RefreshFrequency must be >= 0")
477 }
478
479 // validate the Producer values
480 switch {
481 case c.Producer.MaxMessageBytes <= 0:
482 return ConfigurationError("Producer.MaxMessageBytes must be > 0")
483 case c.Producer.RequiredAcks < -1:
484 return ConfigurationError("Producer.RequiredAcks must be >= -1")
485 case c.Producer.Timeout <= 0:
486 return ConfigurationError("Producer.Timeout must be > 0")
487 case c.Producer.Partitioner == nil:
488 return ConfigurationError("Producer.Partitioner must not be nil")
489 case c.Producer.Flush.Bytes < 0:
490 return ConfigurationError("Producer.Flush.Bytes must be >= 0")
491 case c.Producer.Flush.Messages < 0:
492 return ConfigurationError("Producer.Flush.Messages must be >= 0")
493 case c.Producer.Flush.Frequency < 0:
494 return ConfigurationError("Producer.Flush.Frequency must be >= 0")
495 case c.Producer.Flush.MaxMessages < 0:
496 return ConfigurationError("Producer.Flush.MaxMessages must be >= 0")
497 case c.Producer.Flush.MaxMessages > 0 && c.Producer.Flush.MaxMessages < c.Producer.Flush.Messages:
498 return ConfigurationError("Producer.Flush.MaxMessages must be >= Producer.Flush.Messages when set")
499 case c.Producer.Retry.Max < 0:
500 return ConfigurationError("Producer.Retry.Max must be >= 0")
501 case c.Producer.Retry.Backoff < 0:
502 return ConfigurationError("Producer.Retry.Backoff must be >= 0")
503 }
504
505 if c.Producer.Compression == CompressionLZ4 && !c.Version.IsAtLeast(V0_10_0_0) {
506 return ConfigurationError("lz4 compression requires Version >= V0_10_0_0")
507 }
508
509 if c.Producer.Compression == CompressionGZIP {
510 if c.Producer.CompressionLevel != CompressionLevelDefault {
511 if _, err := gzip.NewWriterLevel(ioutil.Discard, c.Producer.CompressionLevel); err != nil {
512 return ConfigurationError(fmt.Sprintf("gzip compression does not work with level %d: %v", c.Producer.CompressionLevel, err))
513 }
514 }
515 }
516
517 if c.Producer.Idempotent {
518 if !c.Version.IsAtLeast(V0_11_0_0) {
519 return ConfigurationError("Idempotent producer requires Version >= V0_11_0_0")
520 }
521 if c.Producer.Retry.Max == 0 {
522 return ConfigurationError("Idempotent producer requires Producer.Retry.Max >= 1")
523 }
524 if c.Producer.RequiredAcks != WaitForAll {
525 return ConfigurationError("Idempotent producer requires Producer.RequiredAcks to be WaitForAll")
526 }
527 if c.Net.MaxOpenRequests > 1 {
528 return ConfigurationError("Idempotent producer requires Net.MaxOpenRequests to be 1")
529 }
530 }
531
532 // validate the Consumer values
533 switch {
534 case c.Consumer.Fetch.Min <= 0:
535 return ConfigurationError("Consumer.Fetch.Min must be > 0")
536 case c.Consumer.Fetch.Default <= 0:
537 return ConfigurationError("Consumer.Fetch.Default must be > 0")
538 case c.Consumer.Fetch.Max < 0:
539 return ConfigurationError("Consumer.Fetch.Max must be >= 0")
540 case c.Consumer.MaxWaitTime < 1*time.Millisecond:
541 return ConfigurationError("Consumer.MaxWaitTime must be >= 1ms")
542 case c.Consumer.MaxProcessingTime <= 0:
543 return ConfigurationError("Consumer.MaxProcessingTime must be > 0")
544 case c.Consumer.Retry.Backoff < 0:
545 return ConfigurationError("Consumer.Retry.Backoff must be >= 0")
546 case c.Consumer.Offsets.CommitInterval <= 0:
547 return ConfigurationError("Consumer.Offsets.CommitInterval must be > 0")
548 case c.Consumer.Offsets.Initial != OffsetOldest && c.Consumer.Offsets.Initial != OffsetNewest:
549 return ConfigurationError("Consumer.Offsets.Initial must be OffsetOldest or OffsetNewest")
550 case c.Consumer.Offsets.Retry.Max < 0:
551 return ConfigurationError("Consumer.Offsets.Retry.Max must be >= 0")
552 }
553
554 // validate the Consumer Group values
555 switch {
556 case c.Consumer.Group.Session.Timeout <= 2*time.Millisecond:
557 return ConfigurationError("Consumer.Group.Session.Timeout must be >= 2ms")
558 case c.Consumer.Group.Heartbeat.Interval < 1*time.Millisecond:
559 return ConfigurationError("Consumer.Group.Heartbeat.Interval must be >= 1ms")
560 case c.Consumer.Group.Heartbeat.Interval >= c.Consumer.Group.Session.Timeout:
561 return ConfigurationError("Consumer.Group.Heartbeat.Interval must be < Consumer.Group.Session.Timeout")
562 case c.Consumer.Group.Rebalance.Strategy == nil:
563 return ConfigurationError("Consumer.Group.Rebalance.Strategy must not be empty")
564 case c.Consumer.Group.Rebalance.Timeout <= time.Millisecond:
565 return ConfigurationError("Consumer.Group.Rebalance.Timeout must be >= 1ms")
566 case c.Consumer.Group.Rebalance.Retry.Max < 0:
567 return ConfigurationError("Consumer.Group.Rebalance.Retry.Max must be >= 0")
568 case c.Consumer.Group.Rebalance.Retry.Backoff < 0:
569 return ConfigurationError("Consumer.Group.Rebalance.Retry.Backoff must be >= 0")
570 }
571
572 // validate misc shared values
573 switch {
574 case c.ChannelBufferSize < 0:
575 return ConfigurationError("ChannelBufferSize must be >= 0")
576 case !validID.MatchString(c.ClientID):
577 return ConfigurationError("ClientID is invalid")
578 }
579
580 return nil
581}