Add support for SASL and SSL

control-plane/config/200-controller-cluster-role.yaml

@@ -60,6 +60,15 @@ rules:
       - patch
       - watch
 
+  - apiGroups:
+      - "*"
+    resources:
+      - secrets
+    verbs:
+      - list
+      - get
+      - watch
+
   # Eventing resources and statuses we care about
   - apiGroups:
       - "eventing.knative.dev"

control-plane/config/sink/100-kafka-sink.yaml

@@ -82,6 +82,25 @@ spec:
                     - structured
                     - binary
                   default: structured
+                auth:
+                  description: 'Auth configurations'
+                  type: object
+                  properties:
+                    secret:
+                      description: 'Auth secret'
+                      type: object
+                      properties:
+                        ref:
+                          # TODO add format in description (?)
+                          description: |
+                            Secret reference.
+                          type: object
+                          required:
+                            - name
+                          properties:
+                            name:
+                              description: 'Secret name'
+                              type: string
             status:
               description: 'Status represents the current state of the Broker. This data may be out of date.'
               type: object

control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults.go

@@ -27,7 +27,7 @@ func (ks *KafkaSink) SetDefaults(ctx context.Context) {
 func (kss *KafkaSinkSpec) SetDefaults(ctx context.Context) {
 	defaultMode := ModeStructured
 
-	if kss.ContentMode == nil {
+	if kss.ContentMode == nil || *kss.ContentMode == "" {
 		kss.ContentMode = &defaultMode
 	}
 }

control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_types.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
 )
 
 const (
@@ -60,6 +61,7 @@ var _ apis.Defaultable = (*KafkaSink)(nil)
 var _ runtime.Object = (*KafkaSink)(nil)
 var _ duckv1.KRShaped = (*KafkaSink)(nil)
 var _ apis.Convertible = (*KafkaSink)(nil)
+var _ kmeta.OwnerRefable = (*KafkaSink)(nil)
 
 // KafkaSinkSpec defines the desired state of the Kafka Sink.
 type KafkaSinkSpec struct {
@@ -91,6 +93,24 @@ type KafkaSinkSpec struct {
 	//
 	// +optional
 	ContentMode *string `json:"contentMode,omitempty"`
+
+	// Auth configurations.
+	Auth *Auth `json:"auth,omitempty"`
+}
+
+type Auth struct {
+	// Auth Secret
+	Secret *Secret `json:"secret,omitempty"`
+}
+
+type Secret struct {
+	// Secret reference for SASL and SSL configurations.
+	Ref *SecretReference `json:"ref,omitempty"`
+}
+
+type SecretReference struct {
+	// Secret name.
+	Name string `json:"name"`
 }
 
 // KafkaSinkStatus represents the current state of the KafkaSink.
@@ -129,3 +149,7 @@ func (ks *KafkaSink) GetUntypedSpec() interface{} {
 func (ks *KafkaSink) GetStatus() *duckv1.Status {
 	return &ks.Status.Status
 }
+
+func (kss KafkaSinkSpec) HasAuthConfig() bool {
+	return kss.Auth != nil && kss.Auth.Secret != nil && kss.Auth.Secret.Ref != nil
+}

control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation.go

@@ -62,6 +62,10 @@ func (kss *KafkaSinkSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(apis.ErrInvalidValue(*kss.NumPartitions, "numPartitions"))
 	}
 
+	if kss.HasAuthConfig() && kss.Auth.Secret.Ref.Name == "" {
+		errs = errs.Also(apis.ErrInvalidValue("", "auth.secret.ref.name"))
+	}
+
 	return errs
 }
 

control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go

@@ -127,6 +127,19 @@ func TestKafkaSink_Validate(t *testing.T) {
 			ctx:  context.Background(),
 			want: apis.ErrInvalidValue("-10", "spec.replicationFactor"),
 		},
+		{
+			name: "invalid secret name",
+			ks: &KafkaSink{
+				Spec: KafkaSinkSpec{
+					Topic:            "topic-name-1",
+					BootstrapServers: []string{"broker-1:9092"},
+					ContentMode:      pointer.StringPtr(ModeStructured),
+					Auth:             &Auth{Secret: &Secret{Ref: &SecretReference{}}},
+				},
+			},
+			ctx:  context.Background(),
+			want: apis.ErrInvalidValue("", "spec.auth.secret.ref.name"),
+		},
 		{
 			name: "immutable replication factor",
 			ks: &KafkaSink{