From decdac4b6df87f725e8918ff9352f2b80697dfe7 Mon Sep 17 00:00:00 2001
From: Bryan Stitt <bryan@llamanodes.com>
Date: Wed, 19 Apr 2023 10:42:57 -0700
Subject: [PATCH] set protocol

---
 web3_proxy/src/app/mod.rs | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/web3_proxy/src/app/mod.rs b/web3_proxy/src/app/mod.rs
index 6651bffa..243d1313 100644
--- a/web3_proxy/src/app/mod.rs
+++ b/web3_proxy/src/app/mod.rs
@@ -468,9 +468,18 @@ impl Web3ProxyApp {
 
         let mut kafka_producer: Option<rdkafka::producer::FutureProducer> = None;
         if let Some(kafka_brokers) = top_config.app.kafka_urls.clone() {
+            let security_protocol = if kafka_brokers.starts_with("SSL:") {
+                "ssl"
+            } else if kafka_brokers.starts_with("TCP:") {
+                "plaintext"
+            } else {
+                return Err(anyhow::anyhow!("unexpected kafka protocol"));
+            };
+
             match rdkafka::ClientConfig::new()
                 .set("bootstrap.servers", kafka_brokers)
                 .set("message.timeout.ms", "5000")
+                .set("security.protocol", security_protocol)
                 .create()
             {
                 Ok(k) => kafka_producer = Some(k),