allow creating consumer with name

Signed-off-by: Amir Malka <[email protected]>
kubescape · Oct 1, 2024 · 1afdb49 · 1afdb49
1 parent bcb311f
commit 1afdb49
Showing 1 changed file with 8 additions and 0 deletions.
diff --git a/pulsar/connector/consumer.go b/pulsar/connector/consumer.go
@@ -106,6 +106,7 @@ func (c *consumer) applyReconsumeDurationProperties(msg pulsar.Message) {
 }
 
 type createConsumerOptions struct {
+	consumerName         string
 	Topic                TopicName
 	Topics               []TopicName
 	FullTopics           []TopicName
@@ -248,6 +249,12 @@ func WithMessageChannel(messageChannel chan pulsar.ConsumerMessage) CreateConsum
 	}
 }
 
+func WithName(name string) CreateConsumerOption {
+	return func(o *createConsumerOptions) {
+		o.consumerName = name
+	}
+}
+
 func newConsumer(pulsarClient Client, createConsumerOpts ...CreateConsumerOption) (Consumer, error) {
 	opts := &createConsumerOptions{}
 	opts.defaults(pulsarClient.GetConfig())
@@ -294,6 +301,7 @@ func newConsumer(pulsarClient Client, createConsumerOpts ...CreateConsumerOption
 	}
 
 	pulsarConsumer, err := pulsarClient.Subscribe(pulsar.ConsumerOptions{
+		Name:                           opts.consumerName,
 		Topic:                          topic,
 		Topics:                         topics,
 		SubscriptionName:               opts.SubscriptionName,