feat: add replicateSubscriptionState to ConsumerConfig

Add support for replicateSubscriptionState option in ConsumerConfig,
which enables geo-replication failover by synchronizing subscription
cursor state across clusters.

Fixes #478

Author: shibd

index.d.ts

@@ -110,6 +110,7 @@ export interface ConsumerConfig {
   deadLetterPolicy?: DeadLetterPolicy;
   batchReceivePolicy?: ConsumerBatchReceivePolicy;
   keySharedPolicy?: KeySharedPolicy;
+  replicateSubscriptionState?: boolean;
 }
 
 export class Consumer {

src/ConsumerConfig.cc

@@ -62,6 +62,7 @@ static const std::string CFG_KEY_SHARED_POLICY_MODE = "keyShareMode";
 static const std::string CFG_KEY_SHARED_POLICY_ALLOW_OUT_OF_ORDER = "allowOutOfOrderDelivery";
 static const std::string CFG_KEY_SHARED_POLICY_STICKY_RANGES = "stickyRanges";
 static const std::string CFG_CRYPTO_KEY_READER = "cryptoKeyReader";
+static const std::string CFG_REPLICATE_SUBSCRIPTION_STATE = "replicateSubscriptionState";
 
 static const std::map<std::string, pulsar_consumer_type> SUBSCRIPTION_TYPE = {
     {"Exclusive", pulsar_ConsumerExclusive},
@@ -400,6 +401,13 @@ void ConsumerConfig::InitConfig(std::shared_ptr<ThreadSafeDeferred> deferred,
     }
     this->cConsumerConfig.get()->consumerConfiguration.setKeySharedPolicy(cppKeySharedPolicy);
   }
+
+  if (consumerConfig.Has(CFG_REPLICATE_SUBSCRIPTION_STATE) &&
+      consumerConfig.Get(CFG_REPLICATE_SUBSCRIPTION_STATE).IsBoolean()) {
+    bool replicateSubscriptionState = consumerConfig.Get(CFG_REPLICATE_SUBSCRIPTION_STATE).ToBoolean();
+    this->cConsumerConfig.get()->consumerConfiguration.setReplicateSubscriptionStateEnabled(
+        replicateSubscriptionState);
+  }
 }
 
 ConsumerConfig::~ConsumerConfig() {