fix(kafka): support ZooKeeper broker discovery

crates/dbx-core/src/agent_manager.rs

@@ -17,6 +17,10 @@ fn default_jre_key() -> String {
     DEFAULT_JRE_KEY.to_string()
 }
 
+fn strip_utf8_bom(value: &str) -> &str {
+    value.strip_prefix('\u{feff}').unwrap_or(value)
+}
+
 fn is_valid_jar_file(path: &Path) -> bool {
     if !path.is_file() {
         return false;
@@ -138,6 +142,21 @@ mod tests {
         assert!(state.pending_jre_cleanup.is_empty());
     }
 
+    #[test]
+    fn loads_agent_state_with_utf8_bom() {
+        let manager = test_manager("state-utf8-bom");
+        fs::create_dir_all(manager.base_dir()).unwrap();
+        fs::write(
+            manager.state_path(),
+            b"\xEF\xBB\xBF{\"installed_drivers\":{\"kafka\":{\"version\":\"0.1.4\",\"installed_at\":\"now\",\"jre\":\"21\"}}}",
+        )
+        .unwrap();
+
+        let state = manager.load_state();
+
+        assert_eq!(state.installed_drivers.get("kafka").map(|driver| driver.version.as_str()), Some("0.1.4"));
+    }
+
     #[test]
     fn resolves_managed_java_runtime_by_default() {
         let manager = test_manager("managed");
@@ -279,6 +298,25 @@ mod tests {
         );
         assert_eq!(launch.working_dir.as_deref(), Some(driver_dir.as_path()));
     }
+
+    #[test]
+    fn resolves_manifest_agent_launch_with_utf8_bom() {
+        let manager = test_manager("manifest-agent-utf8-bom");
+        let driver_dir = manager.driver_dir("kafka");
+        fs::create_dir_all(&driver_dir).unwrap();
+        fs::write(
+            manager.driver_launch_config_path("kafka"),
+            b"\xEF\xBB\xBF{\"command\":\"java\",\"args\":[\"-jar\",\"{driver_dir}/agent.jar\"]}",
+        )
+        .unwrap();
+
+        let launch = manager
+            .resolve_agent_launch_spec(&AgentState::default(), "kafka", DEFAULT_JRE_KEY)
+            .expect("manifest launch with UTF-8 BOM should resolve");
+
+        assert_eq!(launch.program, PathBuf::from("java"));
+        assert_eq!(launch.args, vec!["-jar".to_string(), format!("{}/agent.jar", driver_dir.to_string_lossy())]);
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -550,7 +588,10 @@ impl AgentManager {
     }
 
     pub fn load_state(&self) -> AgentState {
-        std::fs::read_to_string(self.state_path()).ok().and_then(|s| serde_json::from_str(&s).ok()).unwrap_or_default()
+        std::fs::read_to_string(self.state_path())
+            .ok()
+            .and_then(|s| serde_json::from_str(strip_utf8_bom(&s)).ok())
+            .unwrap_or_default()
     }
 
     pub fn save_state(&self, state: &AgentState) -> Result<(), String> {
@@ -629,7 +670,7 @@ impl AgentManager {
     ) -> Result<AgentLaunchSpec, String> {
         let json = std::fs::read_to_string(config_path)
             .map_err(|e| format!("Failed to read {driver_key} agent launch config: {e}"))?;
-        let config: AgentLaunchConfig = serde_json::from_str(&json)
+        let config: AgentLaunchConfig = serde_json::from_str(strip_utf8_bom(&json))
             .map_err(|e| format!("Failed to parse {driver_key} agent launch config: {e}"))?;
         let command = config.command.trim();
         if command.is_empty() {

crates/dbx-core/src/mq/adapters/kafka.rs

@@ -668,11 +668,13 @@ fn build_connection_params(cfg: &MqAdminConfig) -> serde_json::Value {
     } else {
         "PLAINTEXT"
     });
+    let zookeeper_connect_string = extra_str(extra, "zookeeperServers").unwrap_or("");
     let properties =
         extra.get("properties").filter(|value| value.is_object()).cloned().unwrap_or_else(|| serde_json::json!({}));
 
     serde_json::json!({
         "bootstrap_servers": bootstrap_servers(cfg),
+        "zookeeper_connect_string": zookeeper_connect_string,
         "security_protocol": security_protocol,
         "sasl_mechanism": sasl_mechanism,
         "sasl_username": sasl_username,
@@ -838,6 +840,63 @@ mod tests {
         );
     }
 
+    #[test]
+    fn connection_params_pass_zookeeper_discovery_without_fake_bootstrap_servers() {
+        let cfg = kafka_config(
+            serde_json::json!({
+                "connectionSource": "zookeeper",
+                "zookeeperServers": "zk-1:2181,zk-2:2181/kafka",
+                "securityProtocol": "PLAINTEXT"
+            }),
+            MqAuth::None,
+            false,
+        );
+
+        let params = build_connection_params(&cfg);
+
+        assert_eq!(params.get("bootstrap_servers").and_then(|v| v.as_str()), Some(""));
+        assert_eq!(params.get("zookeeper_connect_string").and_then(|v| v.as_str()), Some("zk-1:2181,zk-2:2181/kafka"));
+        assert_eq!(params.get("security_protocol").and_then(|v| v.as_str()), Some("PLAINTEXT"));
+    }
+
+    #[test]
+    fn connection_params_preserve_zookeeper_sasl_and_tls_properties() {
+        let cfg = kafka_config(
+            serde_json::json!({
+                "connectionSource": "zookeeper",
+                "zookeeperServers": "zk-secure:2281/kafka",
+                "securityProtocol": "SASL_SSL",
+                "properties": {
+                    "zookeeper.sasl.client": "true",
+                    "zookeeper.sasl.clientconfig": "DbxZooKeeperClient",
+                    "zookeeper.client.secure": "true",
+                    "zookeeper.clientCnxnSocket": "org.apache.zookeeper.ClientCnxnSocketNetty",
+                    "zookeeper.ssl.trustStore.location": "/etc/dbx/zookeeper-truststore.p12"
+                }
+            }),
+            MqAuth::None,
+            false,
+        );
+
+        let params = build_connection_params(&cfg);
+
+        assert_eq!(params.get("zookeeper_connect_string").and_then(|v| v.as_str()), Some("zk-secure:2281/kafka"));
+        assert_eq!(params.pointer("/properties/zookeeper.sasl.client").and_then(|v| v.as_str()), Some("true"));
+        assert_eq!(
+            params.pointer("/properties/zookeeper.sasl.clientconfig").and_then(|v| v.as_str()),
+            Some("DbxZooKeeperClient")
+        );
+        assert_eq!(params.pointer("/properties/zookeeper.client.secure").and_then(|v| v.as_str()), Some("true"));
+        assert_eq!(
+            params.pointer("/properties/zookeeper.clientCnxnSocket").and_then(|v| v.as_str()),
+            Some("org.apache.zookeeper.ClientCnxnSocketNetty")
+        );
+        assert_eq!(
+            params.pointer("/properties/zookeeper.ssl.trustStore.location").and_then(|v| v.as_str()),
+            Some("/etc/dbx/zookeeper-truststore.p12")
+        );
+    }
+
     #[test]
     fn reset_cursor_params_preserve_timestamp_position() {
         let topic = TopicRef {