Merge pull request #449 from alibresco/main

Add config option for Kafka version to Kafka output
openconfig · May 22, 2024 · ef4175d · ef4175d
2 parents 0d75e08 + a929f96
commit ef4175d
Showing 1 changed file with 8 additions and 0 deletions.
diff --git a/pkg/outputs/kafka_output/kafka_output.go b/pkg/outputs/kafka_output/kafka_output.go
@@ -95,6 +95,7 @@ type config struct {
 	SplitEvents        bool             `mapstructure:"split-events,omitempty"`
 	NumWorkers         int              `mapstructure:"num-workers,omitempty"`
 	CompressionCodec   string           `mapstructure:"compression-codec,omitempty"`
+	KafkaVersion       string           `mapstructure:"kafka-version,omitempty"`
 	Debug              bool             `mapstructure:"debug,omitempty"`
 	BufferSize         int              `mapstructure:"buffer-size,omitempty"`
 	OverrideTimestamps bool             `mapstructure:"override-timestamps,omitempty"`
@@ -508,6 +509,13 @@ func (k *kafkaOutput) SetTargetsConfig(map[string]*types.TargetConfig) {}
 func (k *kafkaOutput) createConfig() (*sarama.Config, error) {
 	cfg := sarama.NewConfig()
 	cfg.ClientID = k.cfg.Name
+	if k.cfg.KafkaVersion != "" {
+		var err error
+		cfg.Version, err = sarama.ParseKafkaVersion(k.cfg.KafkaVersion)
+		if err != nil {
+			return nil, err
+		}
+	}	
 	// SASL_PLAINTEXT or SASL_SSL
 	if k.cfg.SASL != nil {
 		cfg.Net.SASL.Enable = true