Add --kafka.producer.batch-min-messages collector flag

pkg/kafka/producer/config.go

@@ -36,6 +36,7 @@ type Configuration struct {
 	ProtocolVersion           string                  `mapstructure:"protocol_version"`
 	BatchLinger               time.Duration           `mapstructure:"batch_linger"`
 	BatchSize                 int                     `mapstructure:"batch_size"`
+	BatchMinMessages          int                     `mapstructure:"batch_min_messages"`
 	BatchMaxMessages          int                     `mapstructure:"batch_max_messages"`
 	auth.AuthenticationConfig `mapstructure:"authentication"`
 }
@@ -49,6 +50,7 @@ func (c *Configuration) NewProducer() (sarama.AsyncProducer, error) {
 	saramaConfig.Producer.Return.Successes = true
 	saramaConfig.Producer.Flush.Bytes = c.BatchSize
 	saramaConfig.Producer.Flush.Frequency = c.BatchLinger
+	saramaConfig.Producer.Flush.Messages = c.BatchMinMessages
 	saramaConfig.Producer.Flush.MaxMessages = c.BatchMaxMessages
 	if len(c.ProtocolVersion) > 0 {
 		ver, err := sarama.ParseKafkaVersion(c.ProtocolVersion)

plugin/storage/kafka/options.go

@@ -45,6 +45,7 @@ const (
 	suffixProtocolVersion  = ".protocol-version"
 	suffixBatchLinger      = ".batch-linger"
 	suffixBatchSize        = ".batch-size"
+	suffixBatchMinMessages = ".batch-min-messages"
 	suffixBatchMaxMessages = ".batch-max-messages"
 
 	defaultBroker           = "127.0.0.1:9092"
@@ -55,6 +56,7 @@ const (
 	defaultCompressionLevel = 0
 	defaultBatchLinger      = 0
 	defaultBatchSize        = 0
+	defaultBatchMinMessages = 0
 	defaultBatchMaxMessages = 0
 )
 
@@ -157,10 +159,15 @@ func (opt *Options) AddFlags(flagSet *flag.FlagSet) {
 		defaultBatchSize,
 		"(experimental) Number of bytes to batch before sending records to Kafka. Higher value reduce request to Kafka but increase latency and the possibility of data loss in case of process restart. See https://kafka.apache.org/documentation/",
 	)
+	flagSet.Int(
+		configPrefix+suffixBatchMinMessages,
+		defaultBatchMinMessages,
+		"(experimental) The best-effort minimum number of messages needed to send a batch of records to Kafka. Higher value reduce request to Kafka but increase latency and the possibility of data loss in case of process restart. See https://kafka.apache.org/documentation/",
+	)
 	flagSet.Int(
 		configPrefix+suffixBatchMaxMessages,
 		defaultBatchMaxMessages,
-		"(experimental) Number of message to batch before sending records to Kafka. Higher value reduce request to Kafka but increase latency and the possibility of data loss in case of process restart. See https://kafka.apache.org/documentation/",
+		"(experimental) Maximum number of message to batch before sending records to Kafka",
 	)
 	auth.AddFlags(configPrefix, flagSet)
 }
@@ -195,6 +202,7 @@ func (opt *Options) InitFromViper(v *viper.Viper) {
 		AuthenticationConfig: authenticationOptions,
 		BatchLinger:          v.GetDuration(configPrefix + suffixBatchLinger),
 		BatchSize:            v.GetInt(configPrefix + suffixBatchSize),
+		BatchMinMessages:     v.GetInt(configPrefix + suffixBatchMinMessages),
 		BatchMaxMessages:     v.GetInt(configPrefix + suffixBatchMaxMessages),
 	}
 	opt.Topic = v.GetString(configPrefix + suffixTopic)

plugin/storage/kafka/options_test.go

@@ -40,6 +40,7 @@ func TestOptionsWithFlags(t *testing.T) {
 		"--kafka.producer.compression-level=7",
 		"--kafka.producer.batch-linger=1s",
 		"--kafka.producer.batch-size=128000",
+		"--kafka.producer.batch-min-messages=50",
 		"--kafka.producer.batch-max-messages=100",
 	})
 	opts.InitFromViper(v)
@@ -52,6 +53,7 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, 7, opts.Config.CompressionLevel)
 	assert.Equal(t, 128000, opts.Config.BatchSize)
 	assert.Equal(t, time.Duration(1*time.Second), opts.Config.BatchLinger)
+	assert.Equal(t, 50, opts.Config.BatchMinMessages)
 	assert.Equal(t, 100, opts.Config.BatchMaxMessages)
 }
 
@@ -69,6 +71,7 @@ func TestFlagDefaults(t *testing.T) {
 	assert.Equal(t, 0, opts.Config.CompressionLevel)
 	assert.Equal(t, 0, opts.Config.BatchSize)
 	assert.Equal(t, time.Duration(0*time.Second), opts.Config.BatchLinger)
+	assert.Equal(t, 0, opts.Config.BatchMinMessages)
 	assert.Equal(t, 0, opts.Config.BatchMaxMessages)
 }