Merge pull request #7823 from influxdata/dom/replication-help

refactor(cli): update replication help text

clap_blocks/src/router.rs

@@ -111,16 +111,20 @@ pub struct RouterConfig {
     )]
     pub rpc_write_max_outgoing_bytes: usize,
 
-    /// Specify the optional replication factor for each RPC write.
+    /// Enable optional replication for each RPC write.
     ///
-    /// The total number of copies of data after replication will be this value,
-    /// plus 1.
+    /// This value specifies the total number of copies of data after
+    /// replication, defaulting to 1.
     ///
     /// If the desired replication level is not achieved, a partial write error
     /// will be returned to the user. The write MAY be queryable after a partial
     /// write failure.
-    #[clap(long = "rpc-write-replicas", env = "INFLUXDB_IOX_RPC_WRITE_REPLICAS")]
-    pub rpc_write_replicas: Option<NonZeroUsize>,
+    #[clap(
+        long = "rpc-write-replicas",
+        env = "INFLUXDB_IOX_RPC_WRITE_REPLICAS",
+        default_value = "1"
+    )]
+    pub rpc_write_replicas: NonZeroUsize,
 }
 
 /// Map a string containing an integer number of seconds into a [`Duration`].

influxdb_iox/src/commands/run/all_in_one.rs

@@ -473,7 +473,7 @@ impl Config {
             new_namespace_retention_hours: None, // infinite retention
             namespace_autocreation_enabled: true,
             rpc_write_timeout_seconds: Duration::new(3, 0),
-            rpc_write_replicas: None,
+            rpc_write_replicas: 1.try_into().unwrap(),
             rpc_write_max_outgoing_bytes: ingester_config.rpc_write_max_incoming_bytes,
         };
 

router/src/dml_handlers/rpc_write.rs

@@ -125,7 +125,7 @@ impl<T> RpcWrite<T> {
     /// needed than the number of `endpoints`; doing so will cause a panic.
     pub fn new<N>(
         endpoints: impl IntoIterator<Item = (T, N)>,
-        replica_copies: Option<NonZeroUsize>,
+        n_copies: NonZeroUsize,
         metrics: &metric::Registry,
     ) -> Self
     where
@@ -139,9 +139,9 @@ impl<T> RpcWrite<T> {
             Some(metrics),
         );
 
-        // Map the "replication factor" into the total number of distinct data
-        // copies necessary to consider a write a success.
-        let n_copies = replica_copies.map(NonZeroUsize::get).unwrap_or(1);
+        // Read the total number of distinct data copies necessary to consider a
+        // write a success.
+        let n_copies = n_copies.get();
 
         debug!(n_copies, "write replication factor");
 
@@ -451,7 +451,7 @@ mod tests {
         let client = Arc::new(MockWriteClient::default());
         let handler = RpcWrite::new(
             [(Arc::clone(&client), "mock client")],
-            None,
+            1.try_into().unwrap(),
             &metric::Registry::default(),
         );
 
@@ -513,7 +513,7 @@ mod tests {
                 (Arc::clone(&client2), "client2"),
                 (Arc::clone(&client3), "client3"),
             ],
-            None,
+            1.try_into().unwrap(),
             &metric::Registry::default(),
         );
 
@@ -581,7 +581,7 @@ mod tests {
                 (Arc::clone(&client1), "client1"),
                 (Arc::clone(&client2), "client2"),
             ],
-            None,
+            1.try_into().unwrap(),
             &metric::Registry::default(),
         );
 

router/tests/common/mod.rs

@@ -135,7 +135,11 @@ impl TestContext {
         metrics: Arc<metric::Registry>,
     ) -> Self {
         let client = Arc::new(MockWriteClient::default());
-        let rpc_writer = RpcWrite::new([(Arc::clone(&client), "mock client")], None, &metrics);
+        let rpc_writer = RpcWrite::new(
+            [(Arc::clone(&client), "mock client")],
+            1.try_into().unwrap(),
+            &metrics,
+        );
 
         let ns_cache = Arc::new(ReadThroughCache::new(
             Arc::new(ShardedCache::new(