Simplify Kafka config to use properties file pattern

- Remove --metering-kafka-brokers CLI arg (use properties file instead)
- Make --metering-kafka-properties-file the only required arg for Kafka
- Default --metering-kafka-topic to tips-ingress
- Make --metering-kafka-group-id optional (overrides properties file)
- Load all rdkafka settings from properties file instead of hardcoding

Author: niran

bin/node/src/cli.rs

@@ -46,22 +46,20 @@ pub struct Args {
     pub enable_metering: bool,
 
     // --- Priority fee estimation args ---
-    /// Kafka brokers for metering bundle events (comma-separated)
-    #[arg(long = "metering-kafka-brokers")]
-    pub metering_kafka_brokers: Option<String>,
+    /// Path to Kafka properties file (required for priority fee estimation).
+    /// The properties file should contain rdkafka settings like bootstrap.servers,
+    /// group.id, session.timeout.ms, etc.
+    #[arg(long = "metering-kafka-properties-file")]
+    pub metering_kafka_properties_file: Option<String>,
 
     /// Kafka topic for accepted bundle events
-    #[arg(long = "metering-kafka-topic")]
-    pub metering_kafka_topic: Option<String>,
+    #[arg(long = "metering-kafka-topic", default_value = "tips-ingress")]
+    pub metering_kafka_topic: String,
 
-    /// Kafka consumer group ID
+    /// Kafka consumer group ID (overrides group.id in properties file if set)
     #[arg(long = "metering-kafka-group-id")]
     pub metering_kafka_group_id: Option<String>,
 
-    /// Optional path to Kafka properties file
-    #[arg(long = "metering-kafka-properties-file")]
-    pub metering_kafka_properties_file: Option<String>,
-
     /// Gas limit per flashblock for priority fee estimation
     #[arg(long = "metering-gas-limit", default_value = "30000000")]
     pub metering_gas_limit: u64,
@@ -108,20 +106,12 @@ impl From<Args> for BaseNodeConfig {
             max_pending_blocks_depth: args.max_pending_blocks_depth,
         });
 
-        // Build Kafka config if all required fields are present
-        let kafka = match (
-            args.metering_kafka_brokers,
-            args.metering_kafka_topic,
-            args.metering_kafka_group_id,
-        ) {
-            (Some(brokers), Some(topic), Some(group_id)) => Some(KafkaConfig {
-                brokers,
-                topic,
-                group_id,
-                properties_file: args.metering_kafka_properties_file,
-            }),
-            _ => None,
-        };
+        // Build Kafka config if properties file is provided
+        let kafka = args.metering_kafka_properties_file.map(|properties_file| KafkaConfig {
+            properties_file,
+            topic: args.metering_kafka_topic,
+            group_id_override: args.metering_kafka_group_id,
+        });
 
         let metering = MeteringConfig {
             enabled: args.enable_metering,

crates/runner/src/config.rs

@@ -67,16 +67,18 @@ pub struct MeteringConfig {
 }
 
 /// Kafka connection configuration.
+///
+/// All rdkafka settings (bootstrap.servers, group.id, timeouts, etc.) should be
+/// specified in the properties file. The CLI only specifies the path to this file
+/// and the topic name.
 #[derive(Debug, Clone)]
 pub struct KafkaConfig {
-    /// Comma-separated broker addresses.
-    pub brokers: String,
+    /// Path to the Kafka properties file containing rdkafka settings.
+    pub properties_file: String,
     /// Topic name for accepted bundle events.
     pub topic: String,
-    /// Consumer group ID.
-    pub group_id: String,
-    /// Optional path to properties file.
-    pub properties_file: Option<String>,
+    /// Optional consumer group ID override (takes precedence over properties file).
+    pub group_id_override: Option<String>,
 }
 
 /// Resource limits for priority fee estimation.

crates/runner/src/extensions/rpc.rs

@@ -154,7 +154,7 @@ impl BaseNodeExtension for BaseRpcExtension {
             if metering.enabled && metering.kafka.is_none() {
                 warn!(
                     message = "Metering enabled but Kafka not configured",
-                    help = "Priority fee estimation requires --metering-kafka-brokers, --metering-kafka-topic, and --metering-kafka-group-id"
+                    help = "Priority fee estimation requires --metering-kafka-properties-file"
                 );
             }
 
@@ -201,31 +201,34 @@ impl BaseNodeExtension for BaseRpcExtension {
 
             // Spawn Kafka consumer if configured
             if let (Some(runtime), Some(kafka_cfg)) = (&metering_runtime, &metering.kafka) {
-                info!(message = "Starting Kafka consumer for metering");
+                info!(
+                    message = "Starting Kafka consumer for metering",
+                    properties_file = %kafka_cfg.properties_file,
+                    topic = %kafka_cfg.topic
+                );
 
-                let mut client_config = ClientConfig::new();
-                client_config.set("bootstrap.servers", &kafka_cfg.brokers);
-                client_config.set("group.id", &kafka_cfg.group_id);
-                client_config.set("enable.partition.eof", "false");
-                client_config.set("session.timeout.ms", "6000");
-                client_config.set("enable.auto.commit", "true");
-                client_config.set("auto.offset.reset", "earliest");
-
-                if let Some(path) = kafka_cfg.properties_file.as_ref() {
-                    match load_kafka_config_from_file(path) {
-                        Ok(props) => {
-                            for (key, value) in props {
-                                client_config.set(key, value);
-                            }
-                        }
-                        Err(err) => {
-                            warn!(
-                                message = "Failed to load Kafka properties file",
-                                file = %path,
-                                %err
-                            );
-                        }
+                // Load all rdkafka settings from the properties file
+                let props = match load_kafka_config_from_file(&kafka_cfg.properties_file) {
+                    Ok(props) => props,
+                    Err(err) => {
+                        error!(
+                            target: "metering::kafka",
+                            file = %kafka_cfg.properties_file,
+                            %err,
+                            "Failed to load Kafka properties file"
+                        );
+                        return Ok(());
                     }
+                };
+
+                let mut client_config = ClientConfig::new();
+                for (key, value) in props {
+                    client_config.set(key, value);
+                }
+
+                // Apply CLI override for group.id if specified
+                if let Some(group_id) = &kafka_cfg.group_id_override {
+                    client_config.set("group.id", group_id);
                 }
 
                 let tx_sender = runtime.tx_sender.clone();