(executor/kafka) Added an option to allow self-signed certs (ovh#648)

* Add options for allow unsecure certs when using TLS

Signed-off-by: Armadindon <baptisteperrin77@gmail.com>

* Update executors/kafka/kafka.go

---------

Signed-off-by: Armadindon <baptisteperrin77@gmail.com>
Signed-off-by: Ivan Velasco <ivan.velasco@socotra.com>

executors/kafka/README.md

@@ -15,6 +15,7 @@ In your yaml file, you can use:
   - user optional
   - password optional
   - kafka_version optional, default is 0.10.2.0
+  - insecure_tls optional, permit to allow self-signed certificates when using tls
 
   - client_type mandator: producer or consumer
 

executors/kafka/kafka.go

@@ -2,6 +2,7 @@ package kafka
 
 import (
 	"context"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"os"
@@ -58,6 +59,9 @@ type (
 		User               string `json:"user,omitempty" yaml:"user,omitempty"`
 		Password           string `json:"password,omitempty" yaml:"password,omitempty"`
 
+		// TLS Config
+		InsecureTLS bool `json:"insecure_tls,omitempty" yaml:"insecure_tls,omitempty"`
+
 		// ClientType must be "consumer" or "producer"
 		ClientType string `json:"client_type,omitempty" yaml:"clientType,omitempty"`
 
@@ -361,6 +365,9 @@ func (e Executor) consumeMessages(ctx context.Context) ([]Message, []interface{}
 func (e Executor) getKafkaConfig() (*sarama.Config, error) {
 	config := sarama.NewConfig()
 	config.Net.TLS.Enable = e.WithTLS
+	config.Net.TLS.Config = &tls.Config{
+		InsecureSkipVerify: e.InsecureTLS,
+	}
 	config.Net.SASL.Enable = e.WithSASL
 	config.Net.SASL.User = e.User
 	config.Net.SASL.Password = e.Password