sarama/config.go

package sarama

import (
	"crypto/tls"
	"time"
)

// Config is used to pass multiple configuration options to Sarama's constructors.
type Config struct {
	// Net is the namespace for network-level properties used by the Broker, and shared by the Client/Producer/Consumer.
	Net struct {
		MaxOpenRequests int // How many outstanding requests a connection is allowed to have before sending on it blocks (default 5).

		// All three of the below configurations are similar to the `socket.timeout.ms` setting in JVM kafka.
		DialTimeout  time.Duration // How long to wait for the initial connection to succeed before timing out and returning an error (default 30s).
		ReadTimeout  time.Duration // How long to wait for a response before timing out and returning an error (default 30s).
		WriteTimeout time.Duration // How long to wait for a transmit to succeed before timing out and returning an error (default 30s).

		// NOTE: these config values have no compatibility guarantees; they may change when Kafka releases its
		// official TLS support in version 0.9.
		TLS struct {
			Enable bool        // Whether or not to use TLS when connecting to the broker (defaults to false).
			Config *tls.Config // The TLS configuration to use for secure connections if enabled (defaults to nil).
		}

		// KeepAlive specifies the keep-alive period for an active network connection.
		// If zero, keep-alives are disabled. (default is 0: disabled).
		KeepAlive time.Duration
	}

	// Metadata is the namespace for metadata management properties used by the Client, and shared by the Producer/Consumer.
	Metadata struct {
		Retry struct {
			Max     int           // The total number of times to retry a metadata request when the cluster is in the middle of a leader election (default 3).
			Backoff time.Duration // How long to wait for leader election to occur before retrying (default 250ms). Similar to the JVM's `retry.backoff.ms`.
		}
		// How frequently to refresh the cluster metadata in the background. Defaults to 10 minutes.
		// Set to 0 to disable. Similar to `topic.metadata.refresh.interval.ms` in the JVM version.
		RefreshFrequency time.Duration
	}

	// Producer is the namespace for configuration related to producing messages, used by the Producer.
	Producer struct {
		// The maximum permitted size of a message (defaults to 1000000). Should be set equal to or smaller than the broker's `message.max.bytes`.
		MaxMessageBytes int
		// The level of acknowledgement reliability needed from the broker (defaults to WaitForLocal).
		// Equivalent to the `request.required.acks` setting of the JVM producer.
		RequiredAcks RequiredAcks
		// The maximum duration the broker will wait the receipt of the number of RequiredAcks (defaults to 10 seconds).
		// This is only relevant when RequiredAcks is set to WaitForAll or a number > 1. Only supports millisecond resolution,
		// nanoseconds will be truncated. Equivalent to the JVM producer's `request.timeout.ms` setting.
		Timeout time.Duration
		// The type of compression to use on messages (defaults to no compression). Similar to `compression.codec` setting of the JVM producer.
		Compression CompressionCodec
		// Generates partitioners for choosing the partition to send messages to (defaults to hashing the message key).
		// Similar to the `partitioner.class` setting for the JVM producer.
		Partitioner PartitionerConstructor

		// Return specifies what channels will be populated. If they are set to true, you must read from
		// the respective channels to prevent deadlock.
		Return struct {
			// If enabled, successfully delivered messages will be returned on the Successes channel (default disabled).
			Successes bool

			// If enabled, messages that failed to deliver will be returned on the Errors channel, including error (default enabled).
			Errors bool
		}

		// The following config options control how often messages are batched up and sent to the broker. By default,
		// messages are sent as fast as possible, and all messages received while the current batch is in-flight are placed
		// into the subsequent batch.
		Flush struct {
			Bytes     int           // The best-effort number of bytes needed to trigger a flush. Use the global sarama.MaxRequestSize to set a hard upper limit.
			Messages  int           // The best-effort number of messages needed to trigger a flush. Use `MaxMessages` to set a hard upper limit.
			Frequency time.Duration // The best-effort frequency of flushes. Equivalent to `queue.buffering.max.ms` setting of JVM producer.
			// The maximum number of messages the producer will send in a single broker request.
			// Defaults to 0 for unlimited. Similar to `queue.buffering.max.messages` in the JVM producer.
			MaxMessages int
		}

		Retry struct {
			// The total number of times to retry sending a message (default 3).
			// Similar to the `message.send.max.retries` setting of the JVM producer.
			Max int
			// How long to wait for the cluster to settle between retries (default 100ms).
			// Similar to the `retry.backoff.ms` setting of the JVM producer.
			Backoff time.Duration
		}
	}

	// Consumer is the namespace for configuration related to consuming messages, used by the Consumer.
	Consumer struct {
		Retry struct {
			// How long to wait after a failing to read from a partition before trying again (default 2s).
			Backoff time.Duration
		}

		// Fetch is the namespace for controlling how many bytes are retrieved by any given request.
		Fetch struct {
			// The minimum number of message bytes to fetch in a request - the broker will wait until at least this many are available.
			// The default is 1, as 0 causes the consumer to spin when no messages are available. Equivalent to the JVM's `fetch.min.bytes`.
			Min int32
			// The default number of message bytes to fetch from the broker in each request (default 32768). This should be larger than the
			// majority of your messages, or else the consumer will spend a lot of time negotiating sizes and not actually consuming. Similar
			// to the JVM's `fetch.message.max.bytes`.
			Default int32
			// The maximum number of message bytes to fetch from the broker in a single request. Messages larger than this will return
			// ErrMessageTooLarge and will not be consumable, so you must be sure this is at least as large as your largest message.
			// Defaults to 0 (no limit). Similar to the JVM's `fetch.message.max.bytes`. The global `sarama.MaxResponseSize` still applies.
			Max int32
		}
		// The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it
		// returns fewer than that anyways. The default is 250ms, since 0 causes the consumer to spin when no events are available.
		// 100-500ms is a reasonable range for most cases. Kafka only supports precision up to milliseconds; nanoseconds will be truncated.
		// Equivalent to the JVM's `fetch.wait.max.ms`.
		MaxWaitTime time.Duration

		// The maximum amount of time the consumer expects a message takes to process for the user. If writing to the Messages channel
		// takes longer than this, that partition will stop fetching more messages until it can proceed again. Note that, since the
		// Messages channel is buffered, the actual grace time is (MaxProcessingTime * ChanneBufferSize). Defaults to 100ms.
		MaxProcessingTime time.Duration

		// Return specifies what channels will be populated. If they are set to true, you must read from
		// them to prevent deadlock.
		Return struct {
			// If enabled, any errors that occured while consuming are returned on the Errors channel (default disabled).
			Errors bool
		}

		// Offsets specifies configuration for how and when to commit consumed offsets. This currently requires the
		// manual use of an OffsetManager but will eventually be automated.
		Offsets struct {
			// How frequently to commit updated offsets. Defaults to 1s.
			CommitInterval time.Duration

			// The initial offset to use if no offset was previously committed. Should be OffsetNewest or OffsetOldest.
			// Defaults to OffsetNewest.
			Initial int64
		}
	}

	// A user-provided string sent with every request to the brokers for logging, debugging, and auditing purposes.
	// Defaults to "sarama", but you should probably set it to something specific to your application.
	ClientID string
	// The number of events to buffer in internal and external channels. This permits the producer and consumer to
	// continue processing some messages in the background while user code is working, greatly improving throughput.
	// Defaults to 256.
	ChannelBufferSize int
}

// NewConfig returns a new configuration instance with sane defaults.
func NewConfig() *Config {
	c := &Config{}

	c.Net.MaxOpenRequests = 5
	c.Net.DialTimeout = 30 * time.Second
	c.Net.ReadTimeout = 30 * time.Second
	c.Net.WriteTimeout = 30 * time.Second

	c.Metadata.Retry.Max = 3
	c.Metadata.Retry.Backoff = 250 * time.Millisecond
	c.Metadata.RefreshFrequency = 10 * time.Minute

	c.Producer.MaxMessageBytes = 1000000
	c.Producer.RequiredAcks = WaitForLocal
	c.Producer.Timeout = 10 * time.Second
	c.Producer.Partitioner = NewHashPartitioner
	c.Producer.Retry.Max = 3
	c.Producer.Retry.Backoff = 100 * time.Millisecond
	c.Producer.Return.Errors = true

	c.Consumer.Fetch.Min = 1
	c.Consumer.Fetch.Default = 32768
	c.Consumer.Retry.Backoff = 2 * time.Second
	c.Consumer.MaxWaitTime = 250 * time.Millisecond
	c.Consumer.MaxProcessingTime = 100 * time.Millisecond
	c.Consumer.Return.Errors = false
	c.Consumer.Offsets.CommitInterval = 1 * time.Second
	c.Consumer.Offsets.Initial = OffsetNewest

	c.ChannelBufferSize = 256

	return c
}

// Validate checks a Config instance. It will return a
// ConfigurationError if the specified values don't make sense.
func (c *Config) Validate() error {
	// some configuration values should be warned on but not fail completely, do those first
	if c.Net.TLS.Enable == false && c.Net.TLS.Config != nil {
		Logger.Println("Net.TLS is disabled but a non-nil configuration was provided.")
	}
	if c.Producer.RequiredAcks > 1 {
		Logger.Println("Producer.RequiredAcks > 1 is deprecated and will raise an exception with kafka >= 0.8.2.0.")
	}
	if c.Producer.MaxMessageBytes >= int(MaxRequestSize) {
		Logger.Println("Producer.MaxMessageBytes is larger than MaxRequestSize; it will be ignored.")
	}
	if c.Producer.Flush.Bytes >= int(MaxRequestSize) {
		Logger.Println("Producer.Flush.Bytes is larger than MaxRequestSize; it will be ignored.")
	}
	if c.Producer.Timeout%time.Millisecond != 0 {
		Logger.Println("Producer.Timeout only supports millisecond resolution; nanoseconds will be truncated.")
	}
	if c.Consumer.MaxWaitTime < 100*time.Millisecond {
		Logger.Println("Consumer.MaxWaitTime is very low, which can cause high CPU and network usage. See documentation for details.")
	}
	if c.Consumer.MaxWaitTime%time.Millisecond != 0 {
		Logger.Println("Consumer.MaxWaitTime only supports millisecond precision; nanoseconds will be truncated.")
	}
	if c.ClientID == "sarama" {
		Logger.Println("ClientID is the default of 'sarama', you should consider setting it to something application-specific.")
	}

	// validate Net values
	switch {
	case c.Net.MaxOpenRequests <= 0:
		return ConfigurationError("Net.MaxOpenRequests must be > 0")
	case c.Net.DialTimeout <= 0:
		return ConfigurationError("Net.DialTimeout must be > 0")
	case c.Net.ReadTimeout <= 0:
		return ConfigurationError("Net.ReadTimeout must be > 0")
	case c.Net.WriteTimeout <= 0:
		return ConfigurationError("Net.WriteTimeout must be > 0")
	case c.Net.KeepAlive < 0:
		return ConfigurationError("Net.KeepAlive must be >= 0")
	}

	// validate the Metadata values
	switch {
	case c.Metadata.Retry.Max < 0:
		return ConfigurationError("Metadata.Retry.Max must be >= 0")
	case c.Metadata.Retry.Backoff < 0:
		return ConfigurationError("Metadata.Retry.Backoff must be >= 0")
	case c.Metadata.RefreshFrequency < 0:
		return ConfigurationError("Metadata.RefreshFrequency must be >= 0")
	}

	// validate the Producer values
	switch {
	case c.Producer.MaxMessageBytes <= 0:
		return ConfigurationError("Producer.MaxMessageBytes must be > 0")
	case c.Producer.RequiredAcks < -1:
		return ConfigurationError("Producer.RequiredAcks must be >= -1")
	case c.Producer.Timeout <= 0:
		return ConfigurationError("Producer.Timeout must be > 0")
	case c.Producer.Partitioner == nil:
		return ConfigurationError("Producer.Partitioner must not be nil")
	case c.Producer.Flush.Bytes < 0:
		return ConfigurationError("Producer.Flush.Bytes must be >= 0")
	case c.Producer.Flush.Messages < 0:
		return ConfigurationError("Producer.Flush.Messages must be >= 0")
	case c.Producer.Flush.Frequency < 0:
		return ConfigurationError("Producer.Flush.Frequency must be >= 0")
	case c.Producer.Flush.MaxMessages < 0:
		return ConfigurationError("Producer.Flush.MaxMessages must be >= 0")
	case c.Producer.Flush.MaxMessages > 0 && c.Producer.Flush.MaxMessages < c.Producer.Flush.Messages:
		return ConfigurationError("Producer.Flush.MaxMessages must be >= Producer.Flush.Messages when set")
	case c.Producer.Retry.Max < 0:
		return ConfigurationError("Producer.Retry.Max must be >= 0")
	case c.Producer.Retry.Backoff < 0:
		return ConfigurationError("Producer.Retry.Backoff must be >= 0")
	}

	// validate the Consumer values
	switch {
	case c.Consumer.Fetch.Min <= 0:
		return ConfigurationError("Consumer.Fetch.Min must be > 0")
	case c.Consumer.Fetch.Default <= 0:
		return ConfigurationError("Consumer.Fetch.Default must be > 0")
	case c.Consumer.Fetch.Max < 0:
		return ConfigurationError("Consumer.Fetch.Max must be >= 0")
	case c.Consumer.MaxWaitTime < 1*time.Millisecond:
		return ConfigurationError("Consumer.MaxWaitTime must be >= 1ms")
	case c.Consumer.MaxProcessingTime <= 0:
		return ConfigurationError("Consumer.MaxProcessingTime must be > 0")
	case c.Consumer.Retry.Backoff < 0:
		return ConfigurationError("Consumer.Retry.Backoff must be >= 0")
	case c.Consumer.Offsets.CommitInterval <= 0:
		return ConfigurationError("Consumer.Offsets.CommitInterval must be > 0")
	case c.Consumer.Offsets.Initial != OffsetOldest && c.Consumer.Offsets.Initial != OffsetNewest:
		return ConfigurationError("Consumer.Offsets.Initial must be OffsetOldest or OffsetNewest")

	}

	// validate misc shared values
	switch {
	case c.ChannelBufferSize < 0:
		return ConfigurationError("ChannelBufferSize must be >= 0")
	}

	return nil
}