feat: Adding support for token authentication in Pulsar EventSource

eventsources/sources/pulsar/start.go

@@ -97,6 +97,15 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 		TLSValidateHostname:        pulsarEventSource.TLSValidateHostname,
 	}
 
+	if pulsarEventSource.AuthTokenSecret != nil {
+		token, err := common.GetSecretFromVolume(pulsarEventSource.AuthTokenSecret)
+		if err != nil {
+			log.Errorw("failed to get AuthTokenSecret from the volume", "error", err)
+			return err
+		}
+		clientOpt.Authentication = pulsar.NewAuthenticationToken(token)
+	}
+
 	if pulsarEventSource.TLS != nil {
 		log.Info("setting tls auth option...")
 		var clientCertPath, clientKeyPath string

pkg/apis/eventsource/v1alpha1/types.go

@@ -1024,6 +1024,9 @@ type PulsarEventSource struct {
 	// Metadata holds the user defined metadata which will passed along the event payload.
 	// +optional
 	Metadata map[string]string `json:"metadata,omitempty" protobuf:"bytes,10,rep,name=metadata"`
+	// Authentication token for the pulsar client.
+	// +optional
+	AuthTokenSecret *corev1.SecretKeySelector `json:"authTokenSecret,omitempty" protobuf:"bytes,11,opt,name=authTokenSecret"`
 }
 
 // GenericEventSource refers to a generic event source. It can be used to implement a custom event source.