Fixes output.kafka bulks sizes

auditbeat/auditbeat.reference.yml

@@ -670,10 +670,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

filebeat/filebeat.reference.yml

@@ -1397,10 +1397,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

heartbeat/heartbeat.reference.yml

@@ -814,10 +814,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

journalbeat/journalbeat.reference.yml

@@ -610,10 +610,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

libbeat/_meta/config.reference.yml

@@ -558,10 +558,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

libbeat/docs/outputconfig.asciidoc

@@ -1145,7 +1145,11 @@ endif::[]
 
 ===== `bulk_max_size`
 
-The maximum number of events to bulk in a single Kafka request. The default is 2048.
+The "best effort" number of events to bulk in a single Kafka request. Works only when bulk_flush_frequency > 0; The default is 2048.
+
+===== `bulk_flush_frequency`
+
+Duration to wait before sending bulk Kafka request. 0 is no delay. The default is 0.
 
 ===== `timeout`
 

libbeat/outputs/kafka/config.go

@@ -37,26 +37,27 @@ import (
 )
 
 type kafkaConfig struct {
-	Hosts            []string                  `config:"hosts"               validate:"required"`
-	TLS              *tlscommon.Config         `config:"ssl"`
-	Timeout          time.Duration             `config:"timeout"             validate:"min=1"`
-	Metadata         metaConfig                `config:"metadata"`
-	Key              *fmtstr.EventFormatString `config:"key"`
-	Partition        map[string]*common.Config `config:"partition"`
-	KeepAlive        time.Duration             `config:"keep_alive"          validate:"min=0"`
-	MaxMessageBytes  *int                      `config:"max_message_bytes"   validate:"min=1"`
-	RequiredACKs     *int                      `config:"required_acks"       validate:"min=-1"`
-	BrokerTimeout    time.Duration             `config:"broker_timeout"      validate:"min=1"`
-	Compression      string                    `config:"compression"`
-	CompressionLevel int                       `config:"compression_level"`
-	Version          kafka.Version             `config:"version"`
-	BulkMaxSize      int                       `config:"bulk_max_size"`
-	MaxRetries       int                       `config:"max_retries"         validate:"min=-1,nonzero"`
-	ClientID         string                    `config:"client_id"`
-	ChanBufferSize   int                       `config:"channel_buffer_size" validate:"min=1"`
-	Username         string                    `config:"username"`
-	Password         string                    `config:"password"`
-	Codec            codec.Config              `config:"codec"`
+	Hosts              []string                  `config:"hosts"               validate:"required"`
+	TLS                *tlscommon.Config         `config:"ssl"`
+	Timeout            time.Duration             `config:"timeout"             validate:"min=1"`
+	Metadata           metaConfig                `config:"metadata"`
+	Key                *fmtstr.EventFormatString `config:"key"`
+	Partition          map[string]*common.Config `config:"partition"`
+	KeepAlive          time.Duration             `config:"keep_alive"          validate:"min=0"`
+	MaxMessageBytes    *int                      `config:"max_message_bytes"   validate:"min=1"`
+	RequiredACKs       *int                      `config:"required_acks"       validate:"min=-1"`
+	BrokerTimeout      time.Duration             `config:"broker_timeout"      validate:"min=1"`
+	Compression        string                    `config:"compression"`
+	CompressionLevel   int                       `config:"compression_level"`
+	Version            kafka.Version             `config:"version"`
+	BulkMaxSize        int                       `config:"bulk_max_size"`
+	BulkFlushFrequency time.Duration             `config:"bulk_flush_frequency"`
+	MaxRetries         int                       `config:"max_retries"         validate:"min=-1,nonzero"`
+	ClientID           string                    `config:"client_id"`
+	ChanBufferSize     int                       `config:"channel_buffer_size" validate:"min=1"`
+	Username           string                    `config:"username"`
+	Password           string                    `config:"password"`
+	Codec              codec.Config              `config:"codec"`
 }
 
 type metaConfig struct {
@@ -81,10 +82,11 @@ var compressionModes = map[string]sarama.CompressionCodec{
 
 func defaultConfig() kafkaConfig {
 	return kafkaConfig{
-		Hosts:       nil,
-		TLS:         nil,
-		Timeout:     30 * time.Second,
-		BulkMaxSize: 2048,
+		Hosts:              nil,
+		TLS:                nil,
+		Timeout:            30 * time.Second,
+		BulkMaxSize:        2048,
+		BulkFlushFrequency: 0,
 		Metadata: metaConfig{
 			Retry: metaRetryConfig{
 				Max:     3,
@@ -209,6 +211,12 @@ func newSaramaConfig(config *kafkaConfig) (*sarama.Config, error) {
 	// configure per broker go channel buffering
 	k.ChannelBufferSize = config.ChanBufferSize
 
+	// configure bulk size
+	if config.BulkFlushFrequency > 0 {
+		k.Producer.Flush.Messages = config.BulkMaxSize
+		k.Producer.Flush.Frequency = config.BulkFlushFrequency
+	}
+
 	// configure client ID
 	k.ClientID = config.ClientID
 

metricbeat/metricbeat.reference.yml

@@ -1293,10 +1293,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

packetbeat/packetbeat.reference.yml

@@ -1038,10 +1038,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

winlogbeat/winlogbeat.reference.yml

@@ -591,10 +591,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

x-pack/auditbeat/auditbeat.reference.yml

@@ -721,10 +721,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

x-pack/filebeat/filebeat.reference.yml

@@ -1538,10 +1538,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

x-pack/functionbeat/functionbeat.reference.yml

@@ -727,10 +727,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

x-pack/metricbeat/metricbeat.reference.yml

@@ -1351,10 +1351,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s

x-pack/winlogbeat/winlogbeat.reference.yml

@@ -603,10 +603,14 @@ output.elasticsearch:
   # until all events are published. The default is 3.
   #max_retries: 3
 
-  # The maximum number of events to bulk in a single Kafka request. The default
-  # is 2048.
+  # The "best effort" number of events to bulk in a single Kafka request. Works 
+  # only when bulk_flush_frequency > 0; The default is 2048.
   #bulk_max_size: 2048
 
+  # Duration to wait before sending bulk Kafka request. 0 is no delay. The default
+  # is 0.
+  #bulk_flush_frequency: 10s
+
   # The number of seconds to wait for responses from the Kafka brokers before
   # timing out. The default is 30s.
   #timeout: 30s