Remove pushed_messages mpsc from kafka transforms (#1531)

shotover · Mar 18, 2024 · 2146fa3 · 2146fa3
1 parent 32eaba7
commit 2146fa3
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 14 deletions.
diff --git a/shotover/src/transforms/kafka/sink_cluster/mod.rs b/shotover/src/transforms/kafka/sink_cluster/mod.rs
@@ -29,7 +29,7 @@ use std::net::SocketAddr;
 use std::sync::atomic::AtomicI64;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::sync::{mpsc, oneshot, RwLock};
+use tokio::sync::{oneshot, RwLock};
 use tokio::time::timeout;
 use uuid::Uuid;
 
@@ -158,7 +158,6 @@ impl TransformBuilder for KafkaSinkClusterBuilder {
  Box::new(KafkaSinkCluster {
  first_contact_points: self.first_contact_points.clone(),
  shotover_nodes: self.shotover_nodes.clone(),
- pushed_messages_tx: None,
  read_timeout: self.read_timeout,
  nodes: vec![],
  nodes_shared: self.nodes_shared.clone(),
@@ -236,7 +235,6 @@ impl SaslStatus {
 pub struct KafkaSinkCluster {
  first_contact_points: Vec<String>,
  shotover_nodes: Vec<ShotoverNode>,
- pushed_messages_tx: Option<mpsc::UnboundedSender<Messages>>,
  read_timeout: Option<Duration>,
  nodes: Vec<KafkaNode>,
  nodes_shared: Arc<RwLock<Vec<KafkaNode>>>,
@@ -302,10 +300,6 @@ impl Transform for KafkaSinkCluster {
  self.receive_responses(&find_coordinator_requests, responses)
  .await
  }
-
- fn set_pushed_messages_tx(&mut self, pushed_messages_tx: mpsc::UnboundedSender<Messages>) {
- self.pushed_messages_tx = Some(pushed_messages_tx);
- }
 }
 
 impl KafkaSinkCluster {

diff --git a/shotover/src/transforms/kafka/sink_single.rs b/shotover/src/transforms/kafka/sink_single.rs
@@ -15,7 +15,7 @@ use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use std::time::Duration;
 use tokio::io::split;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::oneshot;
 use tokio::time::timeout;
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -82,7 +82,6 @@ impl TransformBuilder for KafkaSinkSingleBuilder {
  Box::new(KafkaSinkSingle {
  outbound: None,
  address_port: self.address_port,
- pushed_messages_tx: None,
  connect_timeout: self.connect_timeout,
  tls: self.tls.clone(),
  read_timeout: self.read_timeout,
@@ -101,7 +100,6 @@ impl TransformBuilder for KafkaSinkSingleBuilder {
 pub struct KafkaSinkSingle {
  address_port: u16,
  outbound: Option<Connection>,
- pushed_messages_tx: Option<mpsc::UnboundedSender<Messages>>,
  connect_timeout: Duration,
  read_timeout: Option<Duration>,
  tls: Option<TlsConnector>,
@@ -193,10 +191,6 @@ impl Transform for KafkaSinkSingle {
 
  Ok(responses)
  }
-
- fn set_pushed_messages_tx(&mut self, pushed_messages_tx: mpsc::UnboundedSender<Messages>) {
- self.pushed_messages_tx = Some(pushed_messages_tx);
- }
 }
 
 impl KafkaSinkSingle {