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