Add endpoint option for sasl

internal/endpoint/endpoint.go

@@ -66,6 +66,7 @@ type Endpoint struct {
 		Host       string
 		Port       int
 		TopicName  string
+		SASL       bool
 		TLS        bool
 		CACertFile string
 		CertFile   string
@@ -423,6 +424,8 @@ func parseEndpoint(s string) (Endpoint, error) {
 					endpoint.Kafka.CertFile = val[0]
 				case "key":
 					endpoint.Kafka.KeyFile = val[0]
+				case "sasl":
+					endpoint.Kafka.SASL, _ = strconv.ParseBool(val[0])
 				}
 			}
 		}

internal/endpoint/kafka.go

@@ -76,7 +76,9 @@ func (conn *KafkaConn) Send(msg string) error {
 			}
 			cfg.Net.TLS.Enable = true
 			cfg.Net.TLS.Config = tlsConfig
-		} else {
+		}
+
+		if conn.ep.Kafka.SASL {
 			log.Debugf("building kafka sasl config")
 			cfg.ClientID = "sasl_scram_client"
 			cfg.Net.SASL.Enable = true