feat: Add support for replicateSubscriptionState in Pulsar pubsub component

pubsub/pulsar/metadata.go

@@ -39,6 +39,7 @@ type pulsarMetadata struct {
 	ReceiverQueueSize                int                       `mapstructure:"receiverQueueSize"`
 	SubscriptionType                 string                    `mapstructure:"subscribeType"`
 	SubscriptionInitialPosition      string                    `mapstructure:"subscribeInitialPosition"`
+	ReplicateSubscriptionState       bool                      `mapstructure:"replicateSubscriptionState"`
 	Token                            string                    `mapstructure:"token"`
 	oauth2.ClientCredentialsMetadata `mapstructure:",squash"`
 }

pubsub/pulsar/metadata.yaml

@@ -201,4 +201,15 @@ metadata:
     example: '"earliest"'
     url: 
       title: "Pulsar SubscriptionInitialPosition"
-      url: "https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionInitialPosition"
+      url: "https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionInitialPosition"
+  - name: replicateSubscriptionState
+    type: bool
+    description: |
+      Enable replication of subscription state across geo-replicated Pulsar clusters.
+      When enabled, subscription state (such as cursor positions and acknowledgments) will be replicated to other clusters in a geo-replicated setup.
+      This is useful for maintaining subscription consistency during cluster failovers.
+    default: 'false'
+    example: '"true", "false"'
+    url:
+      title: "Pulsar Geo-Replication"
+      url: "https://pulsar.apache.org/docs/administration-geo/"

pubsub/pulsar/pulsar.go

@@ -477,6 +477,7 @@ func (p *Pulsar) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, han
 		MessageChannel:              channel,
 		NackRedeliveryDelay:         p.metadata.RedeliveryDelay,
 		ReceiverQueueSize:           p.metadata.ReceiverQueueSize,
+		ReplicateSubscriptionState:  p.metadata.ReplicateSubscriptionState,
 	}
 
 	// Handle KeySharedPolicy for key_shared subscription type

pubsub/pulsar/pulsar_test.go

@@ -449,6 +449,48 @@ func TestEncryptionKeys(t *testing.T) {
 	})
 }
 
+func TestParsePulsarMetadataReplicateSubscriptionState(t *testing.T) {
+	tt := []struct {
+		name                       string
+		replicateSubscriptionState string
+		expected                   bool
+	}{
+		{
+			name:                       "test replicateSubscriptionState true",
+			replicateSubscriptionState: "true",
+			expected:                   true,
+		},
+		{
+			name:                       "test replicateSubscriptionState false",
+			replicateSubscriptionState: "false",
+			expected:                   false,
+		},
+		{
+			name:                       "test replicateSubscriptionState empty (defaults to false)",
+			replicateSubscriptionState: "",
+			expected:                   false,
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			m := pubsub.Metadata{}
+			m.Properties = map[string]string{
+				"host": "a",
+			}
+
+			if tc.replicateSubscriptionState != "" {
+				m.Properties["replicateSubscriptionState"] = tc.replicateSubscriptionState
+			}
+
+			meta, err := parsePulsarMetadata(m)
+
+			require.NoError(t, err)
+			assert.Equal(t, tc.expected, meta.ReplicateSubscriptionState)
+		})
+	}
+}
+
 func TestSanitiseURL(t *testing.T) {
 	tests := []struct {
 		name     string