feat: support replicate_subscription_state option for consumer

Author: Xanthorrhizol

src/connection.rs

@@ -1639,6 +1639,7 @@ pub(crate) mod messages {
                         .collect(),
                     read_compacted: Some(options.read_compacted.unwrap_or(false)),
                     initial_position: Some(options.initial_position.into()),
+                    replicate_subscription_state: options.replicate_subscription_state,
                     schema: options.schema,
                     start_message_id: options.start_message_id,
                     ..Default::default()

src/consumer/options.rs

@@ -31,6 +31,10 @@ pub struct ConsumerOptions {
     /// }
     /// ```
     pub initial_position: InitialPosition,
+    /// Mark the subscription as "replicated". Pulsar will make sure
+    /// to periodically sync the state of replicated subscriptions
+    /// across different clusters (when using geo-replication).
+    pub replicate_subscription_state: Option<bool>,
 }
 
 impl ConsumerOptions {