fix: sarama version

Signed-off-by: Yvonnick Esnault <[email protected]>

engine/api/event/event.go

@@ -42,7 +42,7 @@ func getBroker(ctx context.Context, t string, option interface{}) (Broker, error
 		k := &KafkaClient{}
 		return k.initialize(ctx, option)
 	}
-	return nil, fmt.Errorf("Invalid Broker Type %s", t)
+	return nil, fmt.Errorf("invalid Broker Type %s", t)
 }
 
 // ResetPublicIntegrations load all integration of type Event and creates kafka brokers

engine/api/event/kafka.go

@@ -24,6 +24,7 @@ type KafkaConfig struct {
 	BrokerAddresses string
 	User            string
 	Password        string
+	Version         string
 	Topic           string
 	MaxMessageByte  int
 	DisableTLS      bool
@@ -35,7 +36,7 @@ type KafkaConfig struct {
 func (c *KafkaClient) initialize(ctx context.Context, options interface{}) (Broker, error) {
 	conf, ok := options.(KafkaConfig)
 	if !ok {
-		return nil, fmt.Errorf("Invalid Kafka Initialization")
+		return nil, fmt.Errorf("invalid Kafka Initialization")
 	}
 
 	if conf.BrokerAddresses == "" ||
@@ -70,6 +71,9 @@ func (c *KafkaClient) initProducer() error {
 		config.Net.SASL.User = c.options.User
 		config.Net.SASL.Password = c.options.Password
 	}
+	if c.options.Version == "" {
+		config.Version = sarama.V0_10_2_0
+	}
 
 	config.ClientID = c.options.ClientID
 	config.Producer.Return.Successes = true