feat: kafka sasl auth

eventsources/sources/kafka/start.go

@@ -270,6 +270,30 @@ func getSaramaConfig(kafkaEventSource *v1alpha1.KafkaEventSource, log *zap.Sugar
 		config.Version = version
 	}
 
+	if kafkaEventSource.SASL != nil {
+		config.Net.SASL.Enable = true
+
+		if kafkaEventSource.SASL.Mechanism == "" {
+			config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypePlaintext)
+		} else {
+			config.Net.SASL.Mechanism = sarama.SASLMechanism(kafkaEventSource.SASL.Mechanism)
+		}
+
+		user, err := common.GetSecretFromVolume(kafkaEventSource.SASL.User)
+		if err != nil {
+			log.Errorf("Error getting user value from secret: %v", err)
+			return nil, err
+		}
+		config.Net.SASL.User = user
+
+		password, err := common.GetSecretFromVolume(kafkaEventSource.SASL.Password)
+		if err != nil {
+			log.Errorf("Error getting password value from secret: %v", err)
+			return nil, err
+		}
+		config.Net.SASL.Password = password
+	}
+
 	if kafkaEventSource.TLS != nil {
 		tlsConfig, err := common.GetTLSConfig(kafkaEventSource.TLS)
 		if err != nil {

eventsources/sources/kafka/validate.go

@@ -46,5 +46,8 @@ func validate(eventSource *v1alpha1.KafkaEventSource) error {
 	if eventSource.TLS != nil {
 		return apicommon.ValidateTLSConfig(eventSource.TLS)
 	}
+	if eventSource.SASL != nil {
+		return apicommon.ValidateSASLConfig(eventSource.SASL)
+	}
 	return nil
 }

pkg/apis/common/common.go

@@ -128,6 +128,19 @@ type TLSConfig struct {
 	DeprecatedClientKeyPath string `json:"clientKeyPath,omitempty" protobuf:"bytes,6,opt,name=clientKeyPath"`
 }
 
+// SASLConfig refers to SASL configuration for a client
+type SASLConfig struct {
+	// SASLMechanism is the name of the enabled SASL mechanism.
+	// Possible values: OAUTHBEARER, PLAIN (defaults to PLAIN).
+	// +optional
+	Mechanism string `json:"mechanism,omitempty" protobuf:"bytes,1,opt,name=mechanism"`
+	// User is the authentication identity (authcid) to present for
+	// SASL/PLAIN or SASL/SCRAM authentication
+	User *corev1.SecretKeySelector `json:"user,omitempty" protobuf:"bytes,2,opt,name=user"`
+	// Password for SASL/PLAIN authentication
+	Password *corev1.SecretKeySelector `json:"password,omitempty" protobuf:"bytes,3,opt,name=password"`
+}
+
 // Backoff for an operation
 type Backoff struct {
 	// The initial duration in nanoseconds or strings like "1s", "3m"