migrate to kafka properties file

Author: danyalprout

.env.example

@@ -3,10 +3,9 @@ TIPS_INGRESS_ADDRESS=0.0.0.0
 TIPS_INGRESS_PORT=8080
 TIPS_INGRESS_RPC_MEMPOOL=http://localhost:2222
 TIPS_INGRESS_DUAL_WRITE_MEMPOOL=false
-TIPS_INGRESS_KAFKA_BROKERS=localhost:9092
-TIPS_INGRESS_KAFKA_TOPIC=tips-audit
+TIPS_INGRESS_KAFKA_INGRESS_PROPERTIES_FILE=/app/docker/ingress-kafka-properties
+TIPS_INGRESS_KAFKA_INGRESS_TOPIC=tips-ingress-rpc
 TIPS_INGRESS_LOG_LEVEL=info
-TIPS_INGRESS_KAFKA_QUEUE_TOPIC=tips-ingress-rpc
 TIPS_INGRESS_SEND_TRANSACTION_DEFAULT_LIFETIME_SECONDS=10800
 
 # Audit service configuration

crates/ingress-rpc/src/main.rs

@@ -4,6 +4,7 @@ use jsonrpsee::server::Server;
 use op_alloy_network::Optimism;
 use rdkafka::ClientConfig;
 use rdkafka::producer::FutureProducer;
+use std::fs;
 use std::net::IpAddr;
 use tracing::{info, warn};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -35,24 +36,16 @@ struct Config {
     dual_write_mempool: bool,
 
     /// Kafka brokers for publishing mempool events
-    #[arg(long, env = "TIPS_INGRESS_KAFKA_BROKERS")]
-    kafka_brokers: String,
-
-    /// Kafka topic for publishing mempool events
-    #[arg(
-        long,
-        env = "TIPS_INGRESS_KAFKA_TOPIC",
-        default_value = "mempool-events"
-    )]
-    kafka_topic: String,
+    #[arg(long, env = "TIPS_INGRESS_KAFKA_INGRESS_PROPERTIES_FILE")]
+    ingress_kafka_properties: String,
 
     /// Kafka topic for queuing transactions before the DB Writer
     #[arg(
         long,
-        env = "TIPS_INGRESS_KAFKA_QUEUE_TOPIC",
+        env = "TIPS_INGRESS_KAFKA_INGRESS_TOPIC",
         default_value = "tips-ingress-rpc"
     )]
-    queue_topic: String,
+    ingress_topic: String,
 
     #[arg(long, env = "TIPS_INGRESS_LOG_LEVEL", default_value = "info")]
     log_level: String,
@@ -106,12 +99,22 @@ async fn main() -> anyhow::Result<()> {
         .network::<Optimism>()
         .connect_http(config.mempool_url);
 
-    let queue_producer: FutureProducer = ClientConfig::new()
-        .set("bootstrap.servers", &config.kafka_brokers)
-        .set("message.timeout.ms", "5000")
-        .create()?;
+    let kafka_properties = fs::read_to_string(&config.ingress_kafka_properties)?;
+    let mut client_config = ClientConfig::new();
+
+    for line in kafka_properties.lines() {
+        let line = line.trim();
+        if line.is_empty() || line.starts_with('#') {
+            continue;
+        }
+        if let Some((key, value)) = line.split_once('=') {
+            client_config.set(key.trim(), value.trim());
+        }
+    }
+
+    let queue_producer: FutureProducer = client_config.create()?;
 
-    let queue = KafkaQueuePublisher::new(queue_producer, config.queue_topic);
+    let queue = KafkaQueuePublisher::new(queue_producer, config.ingress_topic);
 
     let service = IngressService::new(
         provider,

docker-compose.tips.yml

@@ -8,6 +8,8 @@ services:
       - "8080:8080"
     env_file:
       - .env.docker
+    volumes:
+      - ./docker/ingress-kafka-properties:/app/docker/ingress-kafka-properties:ro
     restart: unless-stopped
 
   audit:

docker/ingress-kafka-properties

@@ -0,0 +1,3 @@
+# Kafka configuration properties for ingress service
+bootstrap.servers=host.docker.internal:9092
+message.timeout.ms=5000