Improve kafka output config

libbeat/docs/outputconfig.asciidoc

@@ -627,6 +627,10 @@ out. The default is 30 (seconds).
 
 The maximum duration a broker will wait for number of required ACKs. The default is 10s.
 
+===== channel_buffer_size
+
+Per Kafka broker number of messages buffered in output pipeline. The default is 256.
+
 ===== keep_alive
 
 The keep-alive period for an active network connection. If 0s, keep-alives are disabled. The default is 0 seconds.

libbeat/outputs/kafka/config.go

@@ -18,15 +18,26 @@ type kafkaConfig struct {
 	RequiredACKs    *int               `config:"required_acks"`
 	BrokerTimeout   time.Duration      `config:"broker_timeout"`
 	Compression     string             `config:"compression"`
-	MaxRetries      *int               `config:"max_retries"`
+	MaxRetries      int                `config:"max_retries"`
 	ClientID        string             `config:"client_id"`
+	ChanBufferSize  int                `config:"channel_buffer_size"`
 }
 
 var (
 	defaultConfig = kafkaConfig{
-		Timeout:     30 * time.Second,
-		Worker:      1,
-		Compression: "gzip",
-		ClientID:    "beats",
+		Hosts:           nil,
+		TLS:             nil,
+		Timeout:         30 * time.Second,
+		Worker:          1,
+		UseType:         false,
+		Topic:           "",
+		KeepAlive:       0,
+		MaxMessageBytes: nil, // use library default
+		RequiredACKs:    nil, // use library default
+		BrokerTimeout:   10 * time.Second,
+		Compression:     "gzip",
+		MaxRetries:      3,
+		ClientID:        "beats",
+		ChanBufferSize:  256,
 	}
 )

libbeat/outputs/kafka/kafka.go

@@ -57,7 +57,12 @@ func (k *kafka) init(cfg *common.Config) error {
 		return err
 	}
 
-	libCfg, retries, err := newKafkaConfig(&config)
+	if err := validateConfig(cfg, &config); err != nil {
+		logp.Err("Kafka configuration invalid: %v", err)
+		return err
+	}
+
+	libCfg, err := newKafkaConfig(&config)
 	if err != nil {
 		return err
 	}
@@ -95,7 +100,7 @@ func (k *kafka) init(cfg *common.Config) error {
 	mode, err := mode.NewAsyncConnectionMode(
 		clients,
 		false,
-		retries, // retry implemented by kafka client
+		config.MaxRetries,
 		libCfg.Producer.Retry.Backoff,
 		libCfg.Net.WriteTimeout,
 		10*time.Second)
@@ -128,9 +133,49 @@ func (k *kafka) BulkPublish(
 	return k.mode.PublishEvents(signal, opts, event)
 }
 
-func newKafkaConfig(config *kafkaConfig) (*sarama.Config, int, error) {
+func validateConfig(raw *common.Config, config *kafkaConfig) error {
+	errf := func(fld, msg string) error {
+		return fmt.Errorf("%v %v", raw.PathOf(fld), msg)
+	}
+
+	if len(config.Hosts) == 0 {
+		return errf("hosts", "is empty")
+	}
+	if config.Timeout <= 0 {
+		return errf("timeout", "must be > 0")
+	}
+	if config.BrokerTimeout <= 0 {
+		return errf("broker_timeout", "must be > 0")
+	}
+	if config.Worker <= 0 {
+		return errf("worker", "must be > 0")
+	}
+	if config.UseType == false && config.Topic == "" {
+		return fmt.Errorf("%v or %v",
+			errf("use_type", "must be true"),
+			errf("topic", "must be set"))
+	}
+	if config.KeepAlive < 0 {
+		return errf("keep_alive", "must be >= 0")
+	}
+	if config.MaxMessageBytes != nil && *config.MaxMessageBytes <= 0 {
+		return errf("max_message_bytes", "must be > 0")
+	}
+	if config.RequiredACKs != nil && *config.RequiredACKs < -1 {
+		return errf("required_acks", "must be >= -1")
+	}
+	if _, ok := compressionModes[strings.ToLower(config.Compression)]; !ok {
+		return errf("compression", fmt.Sprintf("'%v' is not supported", config.Compression))
+	}
+	if config.ChanBufferSize <= 0 {
+		return errf("channel_buffer_size", "must be > 0")
+	}
+
+	return nil
+}
+
+func newKafkaConfig(config *kafkaConfig) (*sarama.Config, error) {
 	k := sarama.NewConfig()
-	modeRetries := 1
 
 	// configure network level properties
 	timeout := config.Timeout
@@ -142,7 +187,7 @@ func newKafkaConfig(config *kafkaConfig) (*sarama.Config, int, error) {
 
 	tls, err := outputs.LoadTLSConfig(config.TLS)
 	if err != nil {
-		return nil, modeRetries, err
+		return nil, err
 	}
 	k.Net.TLS.Enable = tls != nil
 	k.Net.TLS.Config = tls
@@ -160,27 +205,24 @@ func newKafkaConfig(config *kafkaConfig) (*sarama.Config, int, error) {
 
 	compressionMode, ok := compressionModes[strings.ToLower(config.Compression)]
 	if !ok {
-		return nil, modeRetries, fmt.Errorf("Unknown compression mode: %v", config.Compression)
+		return nil, fmt.Errorf("Unknown compression mode: %v", config.Compression)
 	}
 	k.Producer.Compression = compressionMode
 
 	k.Producer.Return.Successes = true // enable return channel for signaling
 	k.Producer.Return.Errors = true
 
-	if config.MaxRetries != nil {
-		retries := *config.MaxRetries
-		if retries < 0 {
-			retries = 10
-			modeRetries = -1
-		}
-		k.Producer.Retry.Max = retries
-	}
+	// have retries being handled by libbeat, disable retries in sarama library
+	k.Producer.Retry.Max = 0
+
+	// configure per broker go channel buffering
+	k.ChannelBufferSize = config.ChanBufferSize
 
 	// configure client ID
 	k.ClientID = config.ClientID
 	if err := k.Validate(); err != nil {
 		logp.Err("Invalid kafka configuration: %v", err)
-		return nil, modeRetries, err
+		return nil, err
 	}
-	return k, modeRetries, nil
+	return k, nil
 }

libbeat/outputs/kafka/kafka_integration_test.go

@@ -43,6 +43,7 @@ func getTestKafkaHost() string {
 func newTestKafkaClient(t *testing.T, topic string) *client {
 
 	hosts := []string{getTestKafkaHost()}
+	t.Logf("host: %v", hosts)
 
 	client, err := newKafkaClient(hosts, topic, false, nil)
 	assert.NoError(t, err)
@@ -53,11 +54,10 @@ func newTestKafkaClient(t *testing.T, topic string) *client {
 func newTestKafkaOutput(t *testing.T, topic string, useType bool) outputs.Outputer {
 
 	config := map[string]interface{}{
-		"hosts":          []string{getTestKafkaHost()},
-		"broker_timeout": "1s",
-		"timeout":        1,
-		"topic":          topic,
-		"use_type":       useType,
+		"hosts":    []string{getTestKafkaHost()},
+		"timeout":  "1s",
+		"topic":    topic,
+		"use_type": useType,
 	}
 
 	cfg, err := common.NewConfigFrom(config)
@@ -85,7 +85,9 @@ func testReadFromKafkaTopic(
 	}()
 
 	partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetOldest)
-	assert.NoError(t, err)
+	if err != nil {
+		t.Fatal(err)
+	}
 	defer func() {
 		partitionConsumer.Close()
 	}()