From d0aaa6c22626ed18905a0bf5a239d9af925540da Mon Sep 17 00:00:00 2001
From: Lucas Kent <rubickent@gmail.com>
Date: Tue, 15 Oct 2024 11:29:19 +1100
Subject: [PATCH] Allow disabling shotover peers check

---
 docs/src/transforms.md                           |  1 +
 shotover-proxy/benches/windsock/kafka/bench.rs   |  2 +-
 .../src/transforms/kafka/sink_cluster/mod.rs     | 16 +++++++++-------
 3 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/docs/src/transforms.md b/docs/src/transforms.md
index ad5879dc4..126fb0e2b 100644
--- a/docs/src/transforms.md
+++ b/docs/src/transforms.md
@@ -297,6 +297,7 @@ If SCRAM authentication against the first kafka broker fails, shotover will term
     # If shotover detects that a peer is down shotover will exclude the down peer from its metadata reports to the client.
     # Each peer is checked in a round robin fashion and this `check_shotover_peers_delay_ms` field defines the milliseconds delay taken before moving onto the next peer to check.
     # If the connection cannot be established within connect_timeout_ms, then the peer is considered down.
+    # If this field is not provided, checking of shotover nodes state will be disabled and no outgoing TCP connections to peers will be made.
     check_shotover_peers_delay_ms: 3000
 
     # When this field is provided TLS is used when connecting to the remote address.
diff --git a/shotover-proxy/benches/windsock/kafka/bench.rs b/shotover-proxy/benches/windsock/kafka/bench.rs
index d71ed660a..4f0a524b4 100644
--- a/shotover-proxy/benches/windsock/kafka/bench.rs
+++ b/shotover-proxy/benches/windsock/kafka/bench.rs
@@ -97,7 +97,7 @@ impl KafkaBench {
             KafkaTopology::Cluster1 | KafkaTopology::Cluster3 => Box::new(KafkaSinkClusterConfig {
                 connect_timeout_ms: 3000,
                 read_timeout: None,
-                check_shotover_peers_delay_ms: 3000,
+                check_shotover_peers_delay_ms: Some(3000),
                 first_contact_points: vec![kafka_address],
                 shotover_nodes: vec![ShotoverNodeConfig {
                     address: host_address.parse().unwrap(),
diff --git a/shotover/src/transforms/kafka/sink_cluster/mod.rs b/shotover/src/transforms/kafka/sink_cluster/mod.rs
index 9e748532d..028d2ddb2 100644
--- a/shotover/src/transforms/kafka/sink_cluster/mod.rs
+++ b/shotover/src/transforms/kafka/sink_cluster/mod.rs
@@ -73,7 +73,7 @@ pub struct KafkaSinkClusterConfig {
     pub local_shotover_broker_id: i32,
     pub connect_timeout_ms: u64,
     pub read_timeout: Option<u64>,
-    pub check_shotover_peers_delay_ms: u64,
+    pub check_shotover_peers_delay_ms: Option<u64>,
     pub tls: Option<TlsConnectorConfig>,
     pub authorize_scram_over_mtls: Option<AuthorizeScramOverMtlsConfig>,
 }
@@ -165,7 +165,7 @@ impl KafkaSinkClusterBuilder {
         rack: StrBytes,
         connect_timeout_ms: u64,
         timeout: Option<u64>,
-        check_shotover_peers_delay_ms: u64,
+        check_shotover_peers_delay_ms: Option<u64>,
         tls: Option<TlsConnector>,
     ) -> Result<KafkaSinkClusterBuilder> {
         let read_timeout = timeout.map(Duration::from_secs);
@@ -176,11 +176,13 @@ impl KafkaSinkClusterBuilder {
             .cloned()
             .collect();
 
-        start_shotover_peers_check(
-            shotover_peers,
-            check_shotover_peers_delay_ms,
-            connect_timeout,
-        );
+        if let Some(check_shotover_peers_delay_ms) = check_shotover_peers_delay_ms {
+            start_shotover_peers_check(
+                shotover_peers,
+                check_shotover_peers_delay_ms,
+                connect_timeout,
+            );
+        }
 
         Ok(KafkaSinkClusterBuilder {
             first_contact_points,