diff --git a/shotover-proxy/benches/windsock/cassandra.rs b/shotover-proxy/benches/windsock/cassandra.rs
index 3da61a33f..eceb452ac 100644
--- a/shotover-proxy/benches/windsock/cassandra.rs
+++ b/shotover-proxy/benches/windsock/cassandra.rs
@@ -1,6 +1,6 @@
 use crate::{
     aws::{Ec2InstanceWithDocker, Ec2InstanceWithShotover, RunningShotover},
-    common::{rewritten_file, Shotover},
+    common::{self, Shotover},
     profilers::{self, CloudProfilerRunner, ProfilerRunner},
 };
 use anyhow::Result;
@@ -28,9 +28,20 @@ use scylla::{
     transport::Compression as ScyllaCompression, Session as ScyllaSession,
     SessionBuilder as ScyllaSessionBuilder,
 };
+use shotover::{
+    config::chain::TransformChainConfig,
+    sources::SourceConfig,
+    transforms::{
+        cassandra::{
+            sink_cluster::{CassandraSinkClusterConfig, ShotoverNode},
+            sink_single::CassandraSinkSingleConfig,
+        },
+        debug::force_parse::DebugForceEncodeConfig,
+        TransformConfig,
+    },
+};
 use std::{
     collections::HashMap,
-    path::Path,
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -41,6 +52,7 @@ use test_helpers::{
 };
 use tokio::sync::mpsc::UnboundedSender;
 use tokio_bin_process::bin_path;
+use uuid::Uuid;
 use windsock::{Bench, BenchParameters, BenchTask, Profiling, Report};
 
 const ROW_COUNT: usize = 1000;
@@ -365,6 +377,71 @@ impl CassandraBench {
             cassandra: 1,
         }
     }
+
+    fn generate_topology_yaml(&self, host_address: String, cassandra_address: String) -> String {
+        let mut transforms = vec![];
+        if let Shotover::ForcedMessageParsed = self.shotover {
+            transforms.push(Box::new(DebugForceEncodeConfig {
+                encode_requests: true,
+                encode_responses: true,
+            }) as Box<dyn TransformConfig>);
+        }
+
+        match self.topology {
+            Topology::Cluster3 => {
+                transforms.push(Box::new(CassandraSinkClusterConfig {
+                    first_contact_points: vec![cassandra_address],
+                    tls: None,
+                    connect_timeout_ms: 3000,
+                    local_shotover_host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a".parse().unwrap(),
+                    read_timeout: None,
+                    shotover_nodes: vec![ShotoverNode {
+                        address: host_address.parse().unwrap(),
+                        data_center: "dc1".to_owned(),
+                        rack: "rack1".to_owned(),
+                        host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a".parse().unwrap(),
+                    }],
+                }));
+            }
+            Topology::Single => {
+                transforms.push(Box::new(CassandraSinkSingleConfig {
+                    address: cassandra_address,
+                    tls: None,
+                    connect_timeout_ms: 3000,
+                    read_timeout: None,
+                }));
+            }
+        }
+
+        common::generate_topology(SourceConfig::Cassandra(
+            shotover::sources::cassandra::CassandraConfig {
+                name: "cassandra".to_owned(),
+                listen_addr: host_address,
+                connection_limit: None,
+                hard_connection_limit: None,
+                tls: None,
+                timeout: None,
+                chain: TransformChainConfig(transforms),
+                transport: None,
+            },
+        ))
+    }
+
+    async fn run_aws_shotover(
+        &self,
+        instance: Arc<Ec2InstanceWithShotover>,
+        cassandra_ip: String,
+    ) -> Option<RunningShotover> {
+        let ip = instance.instance.private_ip().to_string();
+        match self.shotover {
+            Shotover::Standard | Shotover::ForcedMessageParsed => {
+                let topology = self
+                    .generate_topology_yaml(format!("{ip}:9042"), format!("{cassandra_ip}:9042"));
+                Some(instance.run_shotover(&topology).await)
+            }
+            Shotover::None => None,
+        }
+    }
 }
 
 #[async_trait]
@@ -488,12 +565,7 @@ impl Bench for CassandraBench {
 
         let (_, running_shotover) = futures::join!(
             run_aws_cassandra(cassandra_nodes, self.topology.clone()),
-            run_aws_shotover(
-                shotover_instance.clone(),
-                self.shotover,
-                cassandra_ip.clone(),
-                self.topology.clone(),
-            )
+            self.run_aws_shotover(shotover_instance.clone(), cassandra_ip.clone(),)
         );
 
         let destination_ip = if running_shotover.is_some() {
@@ -525,10 +597,8 @@ impl Bench for CassandraBench {
 
         let address = match (&self.topology, &self.shotover) {
             (Topology::Single, Shotover::None) => "127.0.0.1:9043",
-            (Topology::Single, Shotover::Standard) => "127.0.0.1:9042",
             (Topology::Cluster3, Shotover::None) => "172.16.1.2:9044",
-            (Topology::Cluster3, Shotover::Standard) => "127.0.0.1:9042",
-            (_, Shotover::ForcedMessageParsed) => todo!(),
+            (_, Shotover::Standard | Shotover::ForcedMessageParsed) => "127.0.0.1:9042",
         };
         let config_dir = match &self.topology {
             Topology::Single => "tests/test-configs/cassandra/passthrough",
@@ -546,16 +616,26 @@ impl Bench for CassandraBench {
                 panic!("Mocked cassandra database does not provide a clustered mode")
             }
         };
+        let cassandra_address = match &self.topology {
+            Topology::Single => "127.0.0.1:9043".to_owned(),
+            Topology::Cluster3 => "172.16.1.2:9044".to_owned(),
+        };
         let mut profiler = ProfilerRunner::new(self.name(), profiling);
         let shotover = match self.shotover {
-            Shotover::Standard => Some(
-                ShotoverProcessBuilder::new_with_topology(&format!("{config_dir}/topology.yaml"))
-                    .with_bin(bin_path!("shotover-proxy"))
-                    .with_profile(profiler.shotover_profile())
-                    .with_cores(core_count.shotover as u32)
-                    .start()
-                    .await,
-            ),
+            Shotover::Standard => {
+                let topology_contents =
+                    self.generate_topology_yaml(address.to_owned(), cassandra_address);
+                let topology_path = std::env::temp_dir().join(Uuid::new_v4().to_string());
+                std::fs::write(&topology_path, topology_contents).unwrap();
+                Some(
+                    ShotoverProcessBuilder::new_with_topology(topology_path.to_str().unwrap())
+                        .with_bin(bin_path!("shotover-proxy"))
+                        .with_profile(profiler.shotover_profile())
+                        .with_cores(core_count.shotover as u32)
+                        .start()
+                        .await,
+                )
+            }
             Shotover::None => None,
             Shotover::ForcedMessageParsed => todo!(),
         };
@@ -627,39 +707,6 @@ impl Bench for CassandraBench {
         self.operation.run(&session, reporter, parameters).await;
     }
 }
-
-async fn run_aws_shotover(
-    instance: Arc<Ec2InstanceWithShotover>,
-    shotover: Shotover,
-    cassandra_ip: String,
-    topology: Topology,
-) -> Option<RunningShotover> {
-    let config_dir = "tests/test-configs/cassandra/bench";
-    let ip = instance.instance.private_ip().to_string();
-    match shotover {
-        Shotover::Standard | Shotover::ForcedMessageParsed => {
-            let encoded = match shotover {
-                Shotover::Standard => "",
-                Shotover::ForcedMessageParsed => "-encode",
-                Shotover::None => unreachable!(),
-            };
-            let clustered = match topology {
-                Topology::Single => "",
-                Topology::Cluster3 => "-cluster",
-            };
-            let topology = rewritten_file(
-                Path::new(&format!(
-                    "{config_dir}/topology{clustered}{encoded}-cloud.yaml"
-                )),
-                &[("HOST_ADDRESS", &ip), ("CASSANDRA_ADDRESS", &cassandra_ip)],
-            )
-            .await;
-            Some(instance.run_shotover(&topology).await)
-        }
-        Shotover::None => None,
-    }
-}
-
 async fn run_aws_cassandra(nodes: Vec<AwsNodeInfo>, topology: Topology) {
     match topology {
         Topology::Cluster3 => run_aws_cassandra_cluster(nodes).await,
diff --git a/shotover-proxy/benches/windsock/common.rs b/shotover-proxy/benches/windsock/common.rs
index cf505bfea..1fd01728b 100644
--- a/shotover-proxy/benches/windsock/common.rs
+++ b/shotover-proxy/benches/windsock/common.rs
@@ -1,5 +1,4 @@
-use anyhow::Context;
-use std::path::Path;
+use shotover::{config::topology::Topology, sources::SourceConfig};
 
 #[derive(Clone, Copy)]
 pub enum Shotover {
@@ -21,13 +20,10 @@ impl Shotover {
     }
 }
 
-pub async fn rewritten_file(path: &Path, find_replace: &[(&str, &str)]) -> String {
-    let mut text = tokio::fs::read_to_string(path)
-        .await
-        .with_context(|| format!("Failed to read from {path:?}"))
-        .unwrap();
-    for (find, replace) in find_replace {
-        text = text.replace(find, replace);
+pub fn generate_topology(source: SourceConfig) -> String {
+    Topology {
+        sources: vec![source],
     }
-    text
+    .serialize()
+    .unwrap()
 }
diff --git a/shotover-proxy/benches/windsock/kafka.rs b/shotover-proxy/benches/windsock/kafka.rs
index 2a40b1281..50421a38b 100644
--- a/shotover-proxy/benches/windsock/kafka.rs
+++ b/shotover-proxy/benches/windsock/kafka.rs
@@ -1,12 +1,16 @@
 use crate::aws::{Ec2InstanceWithDocker, Ec2InstanceWithShotover};
-use crate::common::{rewritten_file, Shotover};
+use crate::common::{self, Shotover};
 use crate::profilers::{self, CloudProfilerRunner, ProfilerRunner};
-use crate::shotover::shotover_process;
+use crate::shotover::shotover_process_custom_topology;
 use anyhow::Result;
 use async_trait::async_trait;
 use aws_throwaway::Ec2Instance;
 use futures::StreamExt;
-use std::path::Path;
+use shotover::config::chain::TransformChainConfig;
+use shotover::sources::SourceConfig;
+use shotover::transforms::debug::force_parse::DebugForceEncodeConfig;
+use shotover::transforms::kafka::sink_single::KafkaSinkSingleConfig;
+use shotover::transforms::TransformConfig;
 use std::sync::Arc;
 use std::{collections::HashMap, time::Duration};
 use test_helpers::docker_compose::docker_compose;
@@ -36,6 +40,48 @@ impl KafkaBench {
             message_size,
         }
     }
+
+    fn generate_topology_yaml(&self, host_address: String, kafka_address: String) -> String {
+        let mut transforms = vec![];
+        if let Shotover::ForcedMessageParsed = self.shotover {
+            transforms.push(Box::new(DebugForceEncodeConfig {
+                encode_requests: true,
+                encode_responses: true,
+            }) as Box<dyn TransformConfig>);
+        }
+
+        transforms.push(Box::new(KafkaSinkSingleConfig {
+            address: kafka_address,
+            connect_timeout_ms: 3000,
+            read_timeout: None,
+        }));
+
+        common::generate_topology(SourceConfig::Kafka(shotover::sources::kafka::KafkaConfig {
+            name: "kafka".to_owned(),
+            listen_addr: host_address,
+            connection_limit: None,
+            hard_connection_limit: None,
+            tls: None,
+            timeout: None,
+            chain: TransformChainConfig(transforms),
+        }))
+    }
+
+    async fn run_aws_shotover(
+        &self,
+        instance: Arc<Ec2InstanceWithShotover>,
+        kafka_ip: String,
+    ) -> Option<crate::aws::RunningShotover> {
+        let ip = instance.instance.private_ip().to_string();
+        match self.shotover {
+            Shotover::Standard | Shotover::ForcedMessageParsed => {
+                let topology =
+                    self.generate_topology_yaml(format!("{ip}:9092"), format!("{kafka_ip}:9092"));
+                Some(instance.run_shotover(&topology).await)
+            }
+            Shotover::None => None,
+        }
+    }
 }
 
 #[async_trait]
@@ -93,7 +139,7 @@ impl Bench for KafkaBench {
 
         let (_, running_shotover) = futures::join!(
             run_aws_kafka(kafka_instance),
-            run_aws_shotover(shotover_instance, self.shotover, kafka_ip.clone())
+            self.run_aws_shotover(shotover_instance, kafka_ip.clone())
         );
 
         let destination_ip = if running_shotover.is_some() {
@@ -125,13 +171,14 @@ impl Bench for KafkaBench {
 
         let mut profiler = ProfilerRunner::new(self.name(), profiling);
         let shotover = match self.shotover {
-            Shotover::Standard => {
-                Some(shotover_process(&format!("{config_dir}/topology.yaml"), &profiler).await)
+            Shotover::Standard | Shotover::ForcedMessageParsed => {
+                let topology_yaml = self.generate_topology_yaml(
+                    "127.0.0.1:9192".to_owned(),
+                    "127.0.0.1:9092".to_owned(),
+                );
+                Some(shotover_process_custom_topology(&topology_yaml, &profiler).await)
             }
             Shotover::None => None,
-            Shotover::ForcedMessageParsed => Some(
-                shotover_process(&format!("{config_dir}/topology-encode.yaml"), &profiler).await,
-            ),
         };
 
         let broker_address = match self.shotover {
@@ -216,31 +263,6 @@ impl Bench for KafkaBench {
     }
 }
 
-async fn run_aws_shotover(
-    instance: Arc<Ec2InstanceWithShotover>,
-    shotover: Shotover,
-    kafka_ip: String,
-) -> Option<crate::aws::RunningShotover> {
-    let config_dir = "tests/test-configs/kafka/bench";
-    let ip = instance.instance.private_ip().to_string();
-    match shotover {
-        Shotover::Standard | Shotover::ForcedMessageParsed => {
-            let encoded = match shotover {
-                Shotover::Standard => "",
-                Shotover::ForcedMessageParsed => "-encode",
-                Shotover::None => unreachable!(),
-            };
-            let topology = rewritten_file(
-                Path::new(&format!("{config_dir}/topology{encoded}-cloud.yaml")),
-                &[("HOST_ADDRESS", &ip), ("KAFKA_ADDRESS", &kafka_ip)],
-            )
-            .await;
-            Some(instance.run_shotover(&topology).await)
-        }
-        Shotover::None => None,
-    }
-}
-
 async fn run_aws_kafka(instance: Arc<Ec2InstanceWithDocker>) {
     let ip = instance.instance.private_ip().to_string();
     instance
diff --git a/shotover-proxy/benches/windsock/redis.rs b/shotover-proxy/benches/windsock/redis.rs
index 404e5af3c..d024dda17 100644
--- a/shotover-proxy/benches/windsock/redis.rs
+++ b/shotover-proxy/benches/windsock/redis.rs
@@ -1,8 +1,8 @@
 use crate::{
     aws::{Ec2InstanceWithDocker, Ec2InstanceWithShotover, RunningShotover, WindsockAws},
-    common::{rewritten_file, Shotover},
+    common::{self, Shotover},
     profilers::{self, CloudProfilerRunner, ProfilerRunner},
-    shotover::shotover_process,
+    shotover::shotover_process_custom_topology,
 };
 use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
@@ -13,12 +13,21 @@ use fred::{
 };
 use itertools::Itertools;
 use rustls_pemfile::{certs, Item};
+use shotover::{
+    config::chain::TransformChainConfig,
+    sources::SourceConfig,
+    tls::{TlsAcceptorConfig, TlsConnectorConfig},
+    transforms::{
+        debug::force_parse::DebugForceEncodeConfig,
+        redis::{sink_cluster::RedisSinkClusterConfig, sink_single::RedisSinkSingleConfig},
+        TransformConfig,
+    },
+};
 use std::{
     collections::HashMap,
     fs::File,
     io::BufReader,
     net::IpAddr,
-    path::Path,
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -68,6 +77,80 @@ impl RedisBench {
             encryption,
         }
     }
+
+    fn generate_topology_yaml(&self, host_address: String, redis_address: String) -> String {
+        let certs = "tests/test-configs/redis/tls/certs";
+        let tls_connector = match self.encryption {
+            Encryption::Tls => Some(TlsConnectorConfig {
+                certificate_authority_path: format!("{certs}/localhost_CA.crt"),
+                certificate_path: Some(format!("{certs}/localhost.crt")),
+                private_key_path: Some(format!("{certs}/localhost.key")),
+                verify_hostname: true,
+            }),
+            Encryption::None => None,
+        };
+        let tls_acceptor = match self.encryption {
+            Encryption::Tls => Some(TlsAcceptorConfig {
+                certificate_path: format!("{certs}/localhost.crt"),
+                private_key_path: format!("{certs}/localhost.key"),
+                certificate_authority_path: None,
+            }),
+            Encryption::None => None,
+        };
+
+        let mut transforms = vec![];
+        if let Shotover::ForcedMessageParsed = self.shotover {
+            transforms.push(Box::new(DebugForceEncodeConfig {
+                encode_requests: true,
+                encode_responses: true,
+            }) as Box<dyn TransformConfig>);
+        }
+
+        match self.topology {
+            RedisTopology::Cluster3 => {
+                transforms.push(Box::new(RedisSinkClusterConfig {
+                    first_contact_points: vec![redis_address],
+                    direct_destination: None,
+                    tls: tls_connector,
+                    connection_count: None,
+                    connect_timeout_ms: 3000,
+                }));
+            }
+            RedisTopology::Single => {
+                transforms.push(Box::new(RedisSinkSingleConfig {
+                    address: redis_address,
+                    tls: tls_connector,
+                    connect_timeout_ms: 3000,
+                }));
+            }
+        }
+
+        common::generate_topology(SourceConfig::Redis(shotover::sources::redis::RedisConfig {
+            name: "redis".to_owned(),
+            listen_addr: host_address,
+            connection_limit: None,
+            hard_connection_limit: None,
+            tls: tls_acceptor,
+            timeout: None,
+            chain: TransformChainConfig(transforms),
+        }))
+    }
+
+    async fn run_aws_shotover(
+        &self,
+        instance: Arc<Ec2InstanceWithShotover>,
+        redis_ip: String,
+    ) -> Option<RunningShotover> {
+        let ip = instance.instance.private_ip().to_string();
+        match self.shotover {
+            Shotover::Standard | Shotover::ForcedMessageParsed => {
+                let topology =
+                    self.generate_topology_yaml(format!("{ip}:6379"), format!("{redis_ip}:6379"));
+                Some(instance.run_shotover(&topology).await)
+            }
+            Shotover::None => None,
+        }
+    }
 }
 
 #[async_trait]
@@ -145,15 +228,10 @@ impl Bench for RedisBench {
         let redis_ip = redis_instances.private_ips()[0].to_string();
         let shotover_ip = shotover_instance.instance.private_ip().to_string();
 
-        redis_instances.run(self.encryption).await;
-        // unlike other sinks, redis cluster sink needs the redis instance to be already up
-        let running_shotover = run_aws_shotover(
-            shotover_instance.clone(),
-            self.shotover,
-            redis_ip.clone(),
-            self.topology,
-        )
-        .await;
+        let (_, running_shotover) = futures::join!(
+            redis_instances.run(self.encryption),
+            self.run_aws_shotover(shotover_instance.clone(), redis_ip.clone())
+        );
 
         let destination_ip = if running_shotover.is_some() {
             format!("redis://{shotover_ip}")
@@ -185,7 +263,7 @@ impl Bench for RedisBench {
         test_helpers::cert::generate_redis_test_certs();
 
         // rediss:// url is not needed to enable TLS because we overwrite the TLS config later on
-        let address = match (self.topology, self.shotover) {
+        let client_url = match (self.topology, self.shotover) {
             (RedisTopology::Single, Shotover::None) => "redis://127.0.0.1:1111",
             (RedisTopology::Cluster3, Shotover::None) => "redis-cluster://172.16.1.2:6379",
             (
@@ -193,6 +271,10 @@ impl Bench for RedisBench {
                 Shotover::Standard | Shotover::ForcedMessageParsed,
             ) => "redis://127.0.0.1:6379",
         };
+        let redis_address = match self.topology {
+            RedisTopology::Single => "127.0.0.1:1111",
+            RedisTopology::Cluster3 => "172.16.1.2:6379",
+        };
         let config_dir = match (self.topology, self.encryption) {
             (RedisTopology::Single, Encryption::None) => "tests/test-configs/redis/passthrough",
             (RedisTopology::Cluster3, Encryption::None) => {
@@ -204,17 +286,16 @@ impl Bench for RedisBench {
         let _compose = docker_compose(&format!("{config_dir}/docker-compose.yaml"));
         let mut profiler = ProfilerRunner::new(self.name(), profiling);
         let shotover = match self.shotover {
-            Shotover::Standard => {
-                Some(shotover_process(&format!("{config_dir}/topology.yaml"), &profiler).await)
+            Shotover::Standard | Shotover::ForcedMessageParsed => {
+                let topology_yaml = self
+                    .generate_topology_yaml("127.0.0.1:6379".to_owned(), redis_address.to_owned());
+                Some(shotover_process_custom_topology(&topology_yaml, &profiler).await)
             }
-            Shotover::ForcedMessageParsed => Some(
-                shotover_process(&format!("{config_dir}/topology-encode.yaml"), &profiler).await,
-            ),
             Shotover::None => None,
         };
         profiler.run(&shotover).await;
 
-        self.execute_run(address, &parameters).await;
+        self.execute_run(client_url, &parameters).await;
 
         if let Some(shotover) = shotover {
             shotover
@@ -370,38 +451,6 @@ impl BenchTask for BenchTaskRedis {
     }
 }
 
-async fn run_aws_shotover(
-    instance: Arc<Ec2InstanceWithShotover>,
-    shotover: Shotover,
-    redis_ip: String,
-    topology: RedisTopology,
-) -> Option<RunningShotover> {
-    let config_dir = "tests/test-configs/redis/bench";
-    let ip = instance.instance.private_ip().to_string();
-    match shotover {
-        Shotover::Standard | Shotover::ForcedMessageParsed => {
-            let clustered = match topology {
-                RedisTopology::Single => "",
-                RedisTopology::Cluster3 => "-cluster",
-            };
-            let encoded = match shotover {
-                Shotover::Standard => "",
-                Shotover::ForcedMessageParsed => "-encode",
-                Shotover::None => unreachable!(),
-            };
-            let topology = rewritten_file(
-                Path::new(&format!(
-                    "{config_dir}/topology{clustered}{encoded}-cloud.yaml"
-                )),
-                &[("HOST_ADDRESS", &ip), ("REDIS_ADDRESS", &redis_ip)],
-            )
-            .await;
-            Some(instance.run_shotover(&topology).await)
-        }
-        Shotover::None => None,
-    }
-}
-
 enum RedisCluster {
     Single(Arc<Ec2InstanceWithDocker>),
     Cluster3 {
diff --git a/shotover-proxy/benches/windsock/shotover.rs b/shotover-proxy/benches/windsock/shotover.rs
index ea53392f6..0f6c5adc2 100644
--- a/shotover-proxy/benches/windsock/shotover.rs
+++ b/shotover-proxy/benches/windsock/shotover.rs
@@ -1,9 +1,15 @@
 use crate::profilers::ProfilerRunner;
 use test_helpers::shotover_process::ShotoverProcessBuilder;
 use tokio_bin_process::{bin_path, BinProcess};
+use uuid::Uuid;
 
-pub async fn shotover_process(topology_path: &str, profiler: &ProfilerRunner) -> BinProcess {
-    ShotoverProcessBuilder::new_with_topology(topology_path)
+pub async fn shotover_process_custom_topology(
+    topology_contents: &str,
+    profiler: &ProfilerRunner,
+) -> BinProcess {
+    let topology_path = std::env::temp_dir().join(Uuid::new_v4().to_string());
+    std::fs::write(&topology_path, topology_contents).unwrap();
+    ShotoverProcessBuilder::new_with_topology(topology_path.to_str().unwrap())
         .with_bin(bin_path!("shotover-proxy"))
         .with_profile(profiler.shotover_profile())
         .start()
diff --git a/shotover-proxy/tests/lib.rs b/shotover-proxy/tests/lib.rs
index ab1c1b0f5..48568237b 100644
--- a/shotover-proxy/tests/lib.rs
+++ b/shotover-proxy/tests/lib.rs
@@ -1,4 +1,4 @@
-#[cfg(test)]
+#[allow(clippy::single_component_path_imports)]
 use rstest_reuse;
 
 use test_helpers::shotover_process::ShotoverProcessBuilder;
diff --git a/shotover-proxy/tests/test-configs/cassandra/bench/topology-cloud.yaml b/shotover-proxy/tests/test-configs/cassandra/bench/topology-cloud.yaml
deleted file mode 100644
index 0b2bcb060..000000000
--- a/shotover-proxy/tests/test-configs/cassandra/bench/topology-cloud.yaml
+++ /dev/null
@@ -1,9 +0,0 @@
----
-sources:
-  - Cassandra:
-      name: "cassandra"
-      listen_addr: "HOST_ADDRESS:9042"
-      chain:
-        - CassandraSinkSingle:
-            remote_address: "CASSANDRA_ADDRESS:9042"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/cassandra/bench/topology-cluster-cloud.yaml b/shotover-proxy/tests/test-configs/cassandra/bench/topology-cluster-cloud.yaml
deleted file mode 100644
index 10c563872..000000000
--- a/shotover-proxy/tests/test-configs/cassandra/bench/topology-cluster-cloud.yaml
+++ /dev/null
@@ -1,15 +0,0 @@
----
-sources:
-  - Cassandra:
-      name: "cassandra"
-      listen_addr: "HOST_ADDRESS:9042"
-      chain:
-        - CassandraSinkCluster:
-            first_contact_points: ["CASSANDRA_ADDRESS:9042"]
-            local_shotover_host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a"
-            shotover_nodes:
-              - address: "HOST_ADDRESS:9042"
-                data_center: "dc1"
-                rack: "rack1"
-                host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/cassandra/bench/topology-cluster-encode-cloud.yaml b/shotover-proxy/tests/test-configs/cassandra/bench/topology-cluster-encode-cloud.yaml
deleted file mode 100644
index 1d8077d55..000000000
--- a/shotover-proxy/tests/test-configs/cassandra/bench/topology-cluster-encode-cloud.yaml
+++ /dev/null
@@ -1,18 +0,0 @@
----
-sources:
-  - Cassandra:
-      name: "cassandra"
-      listen_addr: "HOST_ADDRESS:9042"
-      chain:
-        - DebugForceEncode:
-            encode_requests: true
-            encode_responses: true
-        - CassandraSinkCluster:
-            first_contact_points: ["CASSANDRA_ADDRESS:9042"]
-            local_shotover_host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a"
-            shotover_nodes:
-              - address: "HOST_ADDRESS:9042"
-                data_center: "dc1"
-                rack: "rack1"
-                host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/cassandra/bench/topology-encode-cloud.yaml b/shotover-proxy/tests/test-configs/cassandra/bench/topology-encode-cloud.yaml
deleted file mode 100644
index e9fa8ec63..000000000
--- a/shotover-proxy/tests/test-configs/cassandra/bench/topology-encode-cloud.yaml
+++ /dev/null
@@ -1,12 +0,0 @@
----
-sources:
-  - Cassandra:
-      name: "cassandra"
-      listen_addr: "HOST_ADDRESS:9042"
-      chain:
-        - DebugForceEncode:
-            encode_requests: true
-            encode_responses: true
-        - CassandraSinkSingle:
-            remote_address: "CASSANDRA_ADDRESS:9042"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/kafka/bench/topology-cloud.yaml b/shotover-proxy/tests/test-configs/kafka/bench/topology-cloud.yaml
deleted file mode 100644
index 8fdcf443b..000000000
--- a/shotover-proxy/tests/test-configs/kafka/bench/topology-cloud.yaml
+++ /dev/null
@@ -1,9 +0,0 @@
----
-sources:
-  - Kafka:
-      name: "kafka"
-      listen_addr: "HOST_ADDRESS:9092"
-      chain:
-        - KafkaSinkSingle:
-            remote_address: "KAFKA_ADDRESS:9092"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/kafka/bench/topology-encode-cloud.yaml b/shotover-proxy/tests/test-configs/kafka/bench/topology-encode-cloud.yaml
deleted file mode 100644
index 2edd8dc9f..000000000
--- a/shotover-proxy/tests/test-configs/kafka/bench/topology-encode-cloud.yaml
+++ /dev/null
@@ -1,12 +0,0 @@
----
-sources:
-  - Kafka:
-      name: "kafka"
-      listen_addr: "HOST_ADDRESS:9092"
-      chain:
-        - DebugForceEncode:
-            encode_requests: true
-            encode_responses: true
-        - KafkaSinkSingle:
-            remote_address: "KAFKA_ADDRESS:9092"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/kafka/bench/topology-encode.yaml b/shotover-proxy/tests/test-configs/kafka/bench/topology-encode.yaml
deleted file mode 100644
index 218d3b508..000000000
--- a/shotover-proxy/tests/test-configs/kafka/bench/topology-encode.yaml
+++ /dev/null
@@ -1,12 +0,0 @@
----
-sources:
-  - Kafka:
-      name: "kafka"
-      listen_addr: "127.0.0.1:9192"
-      chain:
-        - DebugForceEncode:
-            encode_requests: true
-            encode_responses: true
-        - KafkaSinkSingle:
-            remote_address: "127.0.0.1:9092"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/kafka/bench/topology.yaml b/shotover-proxy/tests/test-configs/kafka/bench/topology.yaml
deleted file mode 100644
index bdeae01cb..000000000
--- a/shotover-proxy/tests/test-configs/kafka/bench/topology.yaml
+++ /dev/null
@@ -1,9 +0,0 @@
----
-sources:
-  - Kafka:
-      name: "kafka"
-      listen_addr: "127.0.0.1:9192"
-      chain:
-        - KafkaSinkSingle:
-            remote_address: "127.0.0.1:9092"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/redis/bench/topology-cloud.yaml b/shotover-proxy/tests/test-configs/redis/bench/topology-cloud.yaml
deleted file mode 100644
index 1b5503d04..000000000
--- a/shotover-proxy/tests/test-configs/redis/bench/topology-cloud.yaml
+++ /dev/null
@@ -1,9 +0,0 @@
----
-sources:
-  - Redis:
-      name: "redis"
-      listen_addr: "HOST_ADDRESS:6379"
-      chain:
-        - RedisSinkSingle:
-            remote_address: "REDIS_ADDRESS:6379"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/redis/bench/topology-cluster-cloud.yaml b/shotover-proxy/tests/test-configs/redis/bench/topology-cluster-cloud.yaml
deleted file mode 100644
index 5e7dfb78a..000000000
--- a/shotover-proxy/tests/test-configs/redis/bench/topology-cluster-cloud.yaml
+++ /dev/null
@@ -1,10 +0,0 @@
----
-sources:
-  - Redis:
-      name: "redis"
-      listen_addr: "HOST_ADDRESS:6379"
-      chain:
-        - RedisSinkCluster:
-            first_contact_points:
-              - "REDIS_ADDRESS:6379"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/redis/bench/topology-cluster-encode-cloud.yaml b/shotover-proxy/tests/test-configs/redis/bench/topology-cluster-encode-cloud.yaml
deleted file mode 100644
index 60f301802..000000000
--- a/shotover-proxy/tests/test-configs/redis/bench/topology-cluster-encode-cloud.yaml
+++ /dev/null
@@ -1,13 +0,0 @@
----
-sources:
-  - Redis:
-      name: "redis"
-      listen_addr: "HOST_ADDRESS:6379"
-      chain:
-        - DebugForceEncode:
-            encode_requests: true
-            encode_responses: true
-        - RedisSinkCluster:
-            first_contact_points:
-              - "REDIS_ADDRESS:6379"
-            connect_timeout_ms: 3000
diff --git a/shotover-proxy/tests/test-configs/redis/bench/topology-encode-cloud.yaml b/shotover-proxy/tests/test-configs/redis/bench/topology-encode-cloud.yaml
deleted file mode 100644
index 8928f7b5b..000000000
--- a/shotover-proxy/tests/test-configs/redis/bench/topology-encode-cloud.yaml
+++ /dev/null
@@ -1,12 +0,0 @@
----
-sources:
-  - Redis:
-      name: "redis"
-      listen_addr: "HOST_ADDRESS:6379"
-      chain:
-        - DebugForceEncode:
-            encode_requests: true
-            encode_responses: true
-        - RedisSinkSingle:
-            remote_address: "REDIS_ADDRESS:6379"
-            connect_timeout_ms: 3000
diff --git a/windsock-cloud-docker/src/main.rs b/windsock-cloud-docker/src/main.rs
index 523322fad..b92430433 100644
--- a/windsock-cloud-docker/src/main.rs
+++ b/windsock-cloud-docker/src/main.rs
@@ -60,7 +60,7 @@ fn main() {
     container_bash(&format!(
         r#"cd shotover-proxy;
 source "$HOME/.cargo/env";
-AWS_ACCESS_KEY_ID={access_key_id} AWS_SECRET_ACCESS_KEY={secret_access_key} CARGO_TERM_COLOR=always cargo test --target-dir /target --release --bench windsock --features alpha-transforms -- {args}"#
+AWS_ACCESS_KEY_ID={access_key_id} AWS_SECRET_ACCESS_KEY={secret_access_key} CARGO_TERM_COLOR=always cargo test --target-dir /target --release --bench windsock --features alpha-transforms --features rdkafka-driver-tests -- {args}"#
     ));
 
     // extract windsock results