Matteo Scandolo | 9a2772a | 2018-11-19 14:56:26 -0800 | [diff] [blame] | 1 | package sarama |
| 2 | |
| 3 | import ( |
| 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 | |
| 15 | const defaultClientID = "sarama" |
| 16 | |
| 17 | var validID = regexp.MustCompile(`\A[A-Za-z0-9._-]+\z`) |
| 18 | |
| 19 | // Config is used to pass multiple configuration options to Sarama's constructors. |
| 20 | type 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. |
| 343 | func 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. |
| 395 | func (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 | } |