feat: add configurable consumer name for PulsarConsumerSource (#78)

- Add ConsumerName field to PulsarConsumerSource struct
- Extract getConsumerName() method to handle consumer name logic
- Falls back to os.Hostname() when ConsumerName is not set
- Add unit test to validate consumer name behavior

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude <noreply@anthropic.com>

Author: benjamin99

pkg/source/pulsar.go

@@ -165,6 +165,7 @@ type PulsarConsumerSource struct {
 	PulsarSubscription   string
 	PulsarReplicateState bool
 	PulsarMaxReconnect   *uint
+	ConsumerName         string
 
 	client   pulsar.Client
 	consumer pulsar.Consumer
@@ -173,19 +174,25 @@ type PulsarConsumerSource struct {
 	ackTrackers *ackTrackers
 }
 
-func (p *PulsarConsumerSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error) {
-	host, err := os.Hostname()
+func (p *PulsarConsumerSource) getConsumerName() (string, error) {
+	if p.ConsumerName != "" {
+		return p.ConsumerName, nil
+	}
+	hostname, err := os.Hostname()
 	if err != nil {
-		return nil, err
+		return "", err
 	}
+	return hostname, nil
+}
 
-	p.client, err = pulsar.NewClient(p.PulsarOption)
+func (p *PulsarConsumerSource) Capture(cp cursor.Checkpoint) (changes chan Change, err error) {
+	consumerName, err := p.getConsumerName()
 	if err != nil {
 		return nil, err
 	}
 
 	p.consumer, err = p.client.Subscribe(pulsar.ConsumerOptions{
-		Name:                       host,
+		Name:                       consumerName,
 		Topic:                      p.PulsarTopic,
 		SubscriptionName:           p.PulsarSubscription,
 		ReplicateSubscriptionState: p.PulsarReplicateState,

pkg/source/pulsar_test.go

@@ -2,6 +2,7 @@ package source
 
 import (
 	"context"
+	"os"
 	"strconv"
 	"testing"
 	"time"
@@ -250,3 +251,35 @@ func TestPulsarConsumerSource(t *testing.T) {
 	}
 	src.Stop()
 }
+
+func TestPulsarConsumerSourceGetConsumerName(t *testing.T) {
+	// test with custom consumer name set
+	src := &PulsarConsumerSource{
+		ConsumerName: "custom-test-consumer",
+	}
+
+	name, err := src.getConsumerName()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if name != "custom-test-consumer" {
+		t.Fatalf("expected consumer name 'custom-test-consumer', got '%s'", name)
+	}
+
+	// test with empty consumer name (should fall back to hostname)
+	src2 := &PulsarConsumerSource{}
+
+	name2, err := src2.getConsumerName()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if name2 == "" {
+		t.Fatal("expected non-empty hostname, got empty string")
+	}
+
+	// verify that hostname fallback returns a non-empty string
+	hostname, _ := os.Hostname()
+	if name2 != hostname {
+		t.Fatalf("expected hostname '%s', got '%s'", hostname, name2)
+	}
+}