diff --git a/agents/drivers/kafka/build.gradle b/agents/drivers/kafka/build.gradle
index 63f5e38ab..cbea97ab7 100644
--- a/agents/drivers/kafka/build.gradle
+++ b/agents/drivers/kafka/build.gradle
@@ -1,6 +1,15 @@
 dependencies {
     implementation 'com.google.code.gson:gson:2.12.1'
     implementation 'org.apache.kafka:kafka-clients:3.9.0'
+    implementation('org.apache.zookeeper:zookeeper:3.8.4') {
+        exclude group: 'log4j', module: 'log4j'
+        exclude group: 'org.slf4j', module: 'slf4j-log4j12'
+        // The agent stdout is a JSON-RPC transport; Logback's default appender
+        // writes Kafka/ZooKeeper logs to stdout and corrupts that protocol.
+        exclude group: 'ch.qos.logback', module: 'logback-classic'
+        exclude group: 'ch.qos.logback', module: 'logback-core'
+    }
+    implementation 'io.dropwizard.metrics:metrics-core:4.1.12.1'
     runtimeOnly 'org.slf4j:slf4j-simple:1.7.36'
 }
 
diff --git a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java b/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java
index 2330d0ab2..d546bdbba 100644
--- a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java
+++ b/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java
@@ -13,12 +13,21 @@ import org.apache.kafka.common.resource.PatternType;
 import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.resource.ResourcePatternFilter;
 import org.apache.kafka.common.resource.ResourceType;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.BufferedReader;
 import java.io.InputStreamReader;
+import java.io.PrintStream;
+import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.*;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -29,9 +38,12 @@ import java.util.stream.Collectors;
  */
 public final class KafkaAgent {
 
+    private static final PrintStream JSON_RPC_OUT = System.out;
     private static final Gson GSON = new GsonBuilder().serializeNulls().create();
     private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
     private static final int DEFAULT_SESSION_TIMEOUT_MS = 30_000;
+    private static final int DEFAULT_ZOOKEEPER_CONNECTION_TIMEOUT_MS = 10_000;
+    private static final String ZOOKEEPER_PROPERTY_PREFIX = "zookeeper.";
     private static final Set<String> KERBEROS_SYSTEM_PROPERTY_KEYS = Set.of(
         "java.security.krb5.conf",
         "sun.security.krb5.debug",
@@ -47,26 +59,40 @@ public final class KafkaAgent {
 
     private static AdminClient adminClient;
     private static KafkaProducer<String, byte[]> producer;
+    private static JsonObject activeConnection;
     private static volatile boolean shutdownRequested;
 
     private KafkaAgent() {}
 
+    private static Logger logger() {
+        // Initialize only after main redirects System.out, so any logging backend
+        // that defaults to stdout still cannot write into the JSON-RPC channel.
+        return LoggerHolder.INSTANCE;
+    }
+
+    private static final class LoggerHolder {
+        private static final Logger INSTANCE = LoggerFactory.getLogger(KafkaAgent.class);
+    }
+
     // -----------------------------------------------------------------------
     // Entry point
     // -----------------------------------------------------------------------
 
     public static void main(String[] args) throws Exception {
+        // Keep the original stdout exclusively for JSON-RPC. Redirect accidental
+        // System.out writes from dependencies to stderr so they cannot corrupt the protocol.
+        System.setOut(System.err);
         System.setProperty("org.slf4j.simpleLogger.logFile", "System.err");
-        System.out.println("{\"ready\":true}");
-        System.out.flush();
+        JSON_RPC_OUT.println("{\"ready\":true}");
+        JSON_RPC_OUT.flush();
 
         BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
         while (true) {
             String line = reader.readLine();
             if (line == null) break;
             String response = handleRequest(line);
-            System.out.println(response);
-            System.out.flush();
+            JSON_RPC_OUT.println(response);
+            JSON_RPC_OUT.flush();
             if (shutdownRequested) {
                 System.exit(0);
             }
@@ -92,6 +118,7 @@ public final class KafkaAgent {
             Object result = dispatch(method, params);
             response.add("result", GSON.toJsonTree(result));
         } catch (Exception e) {
+            logger().warn("Kafka Agent request failed: method={}, id={}", method, id, e);
             JsonObject error = new JsonObject();
             error.addProperty("code", -1);
             error.addProperty("message", normalizeErrorMessage(e));
@@ -144,7 +171,7 @@ public final class KafkaAgent {
     }
 
     private static Object connect(JsonObject params) throws Exception {
-        JsonObject conn = connectionObject(params);
+        JsonObject conn = resolveBrokerConnection(connectionObject(params));
         Map<String, String> previousKerberosSystemProperties = applyKerberosSystemProperties(conn);
         AdminClient nextAdmin = null;
         KafkaProducer<String, byte[]> nextProducer = null;
@@ -158,6 +185,7 @@ public final class KafkaAgent {
             applyKerberosSystemProperties(conn);
             adminClient = nextAdmin;
             producer = nextProducer;
+            activeConnection = conn.deepCopy();
             return Collections.singletonMap("ok", true);
         } catch (Exception e) {
             if (nextAdmin != null) {
@@ -172,7 +200,7 @@ public final class KafkaAgent {
     }
 
     private static Object testConnection(JsonObject params) throws Exception {
-        JsonObject conn = connectionObject(params);
+        JsonObject conn = resolveBrokerConnection(connectionObject(params));
         Map<String, String> previousKerberosSystemProperties = applyKerberosSystemProperties(conn);
         AdminClient probe = null;
         try {
@@ -189,14 +217,11 @@ public final class KafkaAgent {
                 probe.describeAcls(AclBindingFilter.ANY)
                     .values().get(timeout, TimeUnit.MILLISECONDS);
             } catch (Exception aclEx) {
-                Throwable cause = aclEx;
-                while (cause != null) {
-                    if (cause.getClass().getSimpleName().contains("SecurityDisabled")
-                        || (cause.getMessage() != null && cause.getMessage().contains("No Authorizer"))) {
-                        aclEnabled = false;
-                        break;
-                    }
-                    cause = cause.getCause();
+                if (isAclDisabledError(aclEx)) {
+                    aclEnabled = false;
+                    logger().debug("Kafka ACL support is disabled by the broker");
+                } else {
+                    logger().warn("Kafka ACL capability probe failed; leaving the capability enabled", aclEx);
                 }
             }
 
@@ -219,6 +244,18 @@ public final class KafkaAgent {
         }
     }
 
+    static boolean isAclDisabledError(Throwable error) {
+        Throwable cause = error;
+        while (cause != null) {
+            if (cause.getClass().getSimpleName().contains("SecurityDisabled")
+                || (cause.getMessage() != null && cause.getMessage().contains("No Authorizer"))) {
+                return true;
+            }
+            cause = cause.getCause();
+        }
+        return false;
+    }
+
     private static void closeClients() {
         if (adminClient != null) {
             adminClient.close(Duration.ofSeconds(5));
@@ -228,6 +265,7 @@ public final class KafkaAgent {
             producer.close(Duration.ofSeconds(5));
             producer = null;
         }
+        activeConnection = null;
         restoreKerberosSystemProperties(BASELINE_KERBEROS_SYSTEM_PROPERTIES);
     }
 
@@ -269,6 +307,160 @@ public final class KafkaAgent {
         return servers;
     }
 
+    static JsonObject resolveBrokerConnection(JsonObject conn) throws Exception {
+        String configured = stringOrEmpty(conn, "bootstrap_servers");
+        if (configured.isBlank()) configured = stringOrEmpty(conn, "bootstrapServers");
+        if (!configured.isBlank()) return conn;
+
+        String connectString = stringOrEmpty(conn, "zookeeper_connect_string");
+        if (connectString.isBlank()) connectString = stringOrEmpty(conn, "zookeeperServers");
+        if (connectString.isBlank()) {
+            throw new IllegalArgumentException("bootstrap_servers or zookeeper_connect_string is required");
+        }
+
+        JsonObject resolved = conn.deepCopy();
+        resolved.addProperty("bootstrap_servers", discoverBootstrapServers(connectString, securityProtocol(conn), conn));
+        return resolved;
+    }
+
+    private static String discoverBootstrapServers(String connectString, String securityProtocol, JsonObject conn)
+        throws Exception {
+        int sessionTimeout = intOrDefault(conn, "zookeeper_session_timeout_ms", DEFAULT_SESSION_TIMEOUT_MS);
+        int connectionTimeout = intOrDefault(
+            conn,
+            "zookeeper_connection_timeout_ms",
+            DEFAULT_ZOOKEEPER_CONNECTION_TIMEOUT_MS
+        );
+        CountDownLatch connected = new CountDownLatch(1);
+        ZooKeeper zooKeeper = new ZooKeeper(connectString, sessionTimeout, event -> {
+            if (event.getState() == Watcher.Event.KeeperState.SyncConnected) connected.countDown();
+        }, zooKeeperClientConfig(conn));
+        try {
+            if (!connected.await(connectionTimeout, TimeUnit.MILLISECONDS)) {
+                throw new IllegalStateException("Timed out connecting to ZooKeeper for Kafka broker discovery");
+            }
+
+            List<String> brokerIds;
+            try {
+                brokerIds = new ArrayList<>(zooKeeper.getChildren("/brokers/ids", false));
+            } catch (KeeperException.NoNodeException e) {
+                throw new IllegalStateException("ZooKeeper path /brokers/ids does not exist", e);
+            }
+            brokerIds.sort(KafkaAgent::compareBrokerIds);
+
+            List<JsonObject> registrations = new ArrayList<>();
+            for (String brokerId : brokerIds) {
+                try {
+                    byte[] data = zooKeeper.getData("/brokers/ids/" + brokerId, false, null);
+                    registrations.add(JsonParser.parseString(new String(data, StandardCharsets.UTF_8)).getAsJsonObject());
+                } catch (KeeperException.NoNodeException e) {
+                    // Expected race: a broker may refresh its ephemeral node between list and read.
+                    logger().debug("Kafka broker {} disappeared during ZooKeeper discovery", brokerId);
+                } catch (RuntimeException e) {
+                    logger().warn("Skipping malformed ZooKeeper registration for Kafka broker {}", brokerId, e);
+                }
+            }
+            return brokerEndpoints(registrations, securityProtocol);
+        } finally {
+            zooKeeper.close();
+        }
+    }
+
+    static ZKClientConfig zooKeeperClientConfig(JsonObject conn) {
+        ZKClientConfig clientConfig = new ZKClientConfig();
+        JsonObject properties = connectionProperties(conn);
+        if (properties == null) return clientConfig;
+
+        for (Map.Entry<String, JsonElement> entry : properties.entrySet()) {
+            if (entry.getKey().startsWith(ZOOKEEPER_PROPERTY_PREFIX)
+                && entry.getValue().isJsonPrimitive()) {
+                clientConfig.setProperty(entry.getKey(), entry.getValue().getAsString());
+            }
+        }
+        return clientConfig;
+    }
+
+    private static int compareBrokerIds(String left, String right) {
+        try {
+            return Integer.compare(Integer.parseInt(left), Integer.parseInt(right));
+        } catch (NumberFormatException e) {
+            // Third-party registries may use non-numeric IDs; lexical ordering remains deterministic.
+            logger().debug("Sorting non-numeric Kafka broker IDs lexically: left={}, right={}", left, right);
+            return left.compareTo(right);
+        }
+    }
+
+    static String brokerEndpoints(List<JsonObject> registrations, String securityProtocol) {
+        String targetProtocol = securityProtocol == null || securityProtocol.isBlank()
+            ? "PLAINTEXT"
+            : securityProtocol.toUpperCase(Locale.ROOT);
+        Set<String> addresses = new LinkedHashSet<>();
+
+        for (JsonObject registration : registrations) {
+            int addressCount = addresses.size();
+            JsonObject protocolMap = registration.has("listener_security_protocol_map")
+                && registration.get("listener_security_protocol_map").isJsonObject()
+                ? registration.getAsJsonObject("listener_security_protocol_map")
+                : new JsonObject();
+            JsonArray endpoints = registration.has("endpoints") && registration.get("endpoints").isJsonArray()
+                ? registration.getAsJsonArray("endpoints")
+                : new JsonArray();
+
+            for (JsonElement element : endpoints) {
+                if (!element.isJsonPrimitive() || !element.getAsJsonPrimitive().isString()) continue;
+                String endpoint = element.getAsString();
+                int separator = endpoint.indexOf("://");
+                if (separator <= 0) continue;
+                String listener = endpoint.substring(0, separator).toUpperCase(Locale.ROOT);
+                JsonElement mapped = protocolMap.get(listener);
+                String mappedProtocol = mapped != null && mapped.isJsonPrimitive()
+                    ? mapped.getAsString().toUpperCase(Locale.ROOT)
+                    : listener;
+                if (!targetProtocol.equals(mappedProtocol)) continue;
+                String address = endpointAddress(endpoint);
+                if (address != null) addresses.add(address);
+            }
+
+            if (addresses.size() == addressCount && endpoints.size() == 0 && registration.has("host") && registration.has("port")) {
+                try {
+                    String host = registration.get("host").getAsString().trim();
+                    int port = registration.get("port").getAsInt();
+                    if (!host.isEmpty() && port > 0 && port <= 65535) addresses.add(formatHostPort(host, port));
+                } catch (RuntimeException e) {
+                    logger().warn("Skipping malformed legacy Kafka broker registration", e);
+                }
+            }
+        }
+
+        if (addresses.isEmpty()) {
+            throw new IllegalArgumentException("ZooKeeper did not return any usable Kafka broker endpoints");
+        }
+        return String.join(",", addresses);
+    }
+
+    private static String endpointAddress(String endpoint) {
+        try {
+            URI uri = URI.create(endpoint);
+            String host = uri.getHost();
+            int port = uri.getPort();
+            if (host == null || host.isBlank() || port <= 0 || port > 65535) return null;
+            return formatHostPort(host, port);
+        } catch (IllegalArgumentException e) {
+            logger().debug("Skipping malformed Kafka broker endpoint", e);
+            return null;
+        }
+    }
+
+    private static String formatHostPort(String host, int port) {
+        return host.contains(":") && !host.startsWith("[") ? "[" + host + "]:" + port : host + ":" + port;
+    }
+
+    private static String securityProtocol(JsonObject conn) {
+        String protocol = stringOrEmpty(conn, "security_protocol");
+        if (protocol.isBlank()) protocol = stringOrEmpty(conn, "securityProtocol");
+        return protocol.isBlank() ? "PLAINTEXT" : protocol;
+    }
+
     static void applySecurityProperties(JsonObject conn, Properties props) {
         String securityProtocol = stringOrEmpty(conn, "security_protocol");
         if (securityProtocol.isBlank()) {
@@ -353,6 +545,7 @@ public final class KafkaAgent {
             for (Map.Entry<String, JsonElement> entry : properties.entrySet()) {
                 if (entry.getValue().isJsonPrimitive()) {
                     String key = entry.getKey();
+                    if (key.startsWith(ZOOKEEPER_PROPERTY_PREFIX)) continue;
                     String value = entry.getValue().getAsString();
                     props.put(key, value);
                 }
@@ -578,11 +771,47 @@ public final class KafkaAgent {
         }
 
         ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, name);
-        admin.incrementalAlterConfigs(Collections.singletonMap(resource, ops))
-            .all().get(timeout, TimeUnit.MILLISECONDS);
+        try {
+            admin.incrementalAlterConfigs(Collections.singletonMap(resource, ops))
+                .all().get(timeout, TimeUnit.MILLISECONDS);
+        } catch (Exception e) {
+            if (!isUnsupportedVersionError(e)) throw e;
+            logger().info("Kafka broker does not support incrementalAlterConfigs; using legacy alterConfigs for topic {}", name);
+            Config current = admin.describeConfigs(Collections.singletonList(resource))
+                .all().get(timeout, TimeUnit.MILLISECONDS).get(resource);
+            Map<String, String> values = legacyTopicConfig(current, ops);
+            Config replacement = new Config(values.entrySet().stream()
+                .map(entry -> new ConfigEntry(entry.getKey(), entry.getValue()))
+                .collect(Collectors.toList()));
+            admin.alterConfigs(Collections.singletonMap(resource, replacement))
+                .all().get(timeout, TimeUnit.MILLISECONDS);
+        }
         return Collections.singletonMap("ok", true);
     }
 
+    static Map<String, String> legacyTopicConfig(Config current, List<AlterConfigOp> ops) {
+        Map<String, String> values = new LinkedHashMap<>();
+        for (ConfigEntry entry : current.entries()) {
+            boolean topicOverride = entry.source() == ConfigEntry.ConfigSource.DYNAMIC_TOPIC_CONFIG
+                || entry.source() == ConfigEntry.ConfigSource.UNKNOWN;
+            if (topicOverride && !entry.isReadOnly() && !entry.isSensitive() && entry.value() != null) {
+                values.put(entry.name(), entry.value());
+            }
+        }
+
+        for (AlterConfigOp op : ops) {
+            String key = op.configEntry().name();
+            switch (op.opType()) {
+                case SET -> values.put(key, op.configEntry().value());
+                case DELETE -> values.remove(key);
+                case APPEND, SUBTRACT -> throw new IllegalArgumentException(
+                    "Kafka broker does not support " + op.opType() + " config operations through the legacy alterConfigs API"
+                );
+            }
+        }
+        return values;
+    }
+
     // -----------------------------------------------------------------------
     // Consumer groups
     // -----------------------------------------------------------------------
@@ -698,6 +927,7 @@ public final class KafkaAgent {
             return Collections.singletonMap("producers", new ArrayList<>(byProducer.values()));
         } catch (Exception e) {
             if (isUnsupportedVersionError(e)) {
+                logger().info("Kafka broker does not support describeProducers; returning an empty producer list");
                 return Collections.singletonMap("producers", Collections.emptyList());
             }
             throw e;
@@ -797,26 +1027,11 @@ public final class KafkaAgent {
         Long offset = longOrNull(params, "offset");
         int count = Math.max(1, intOrDefault(params, "count", 10));
 
-        // Build a temporary consumer for peeking (no commit)
-        Properties props = new Properties();
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
-            adminClient != null ? adminClient.describeCluster().clusterId()
-                .get(5, TimeUnit.SECONDS) : "localhost:9092");
-        // Reuse the admin's bootstrap servers
-        JsonObject conn = params.has("connection") && params.get("connection").isJsonObject()
-            ? params.getAsJsonObject("connection") : null;
-        if (conn != null) {
-            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers(conn));
-            applyConnectionProperties(conn, props);
+        JsonObject conn = activeConnection;
+        if (conn == null) {
+            throw new IllegalStateException("Kafka Agent is not connected");
         }
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, "dbx-peek-" + UUID.randomUUID());
-        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
-            "org.apache.kafka.common.serialization.StringDeserializer");
-        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
-            "org.apache.kafka.common.serialization.ByteArrayDeserializer");
-        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
-        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
-        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, count);
+        Properties props = peekConsumerProperties(conn, count);
 
         try (KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props)) {
             List<TopicPartition> candidatePartitions = resolvePeekPartitions(consumer, topic, partition);
@@ -872,6 +1087,21 @@ public final class KafkaAgent {
         }
     }
 
+    static Properties peekConsumerProperties(JsonObject conn, int count) {
+        Properties props = new Properties();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers(conn));
+        applyConnectionProperties(conn, props);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "dbx-peek-" + UUID.randomUUID());
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+            "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+            "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
+        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, count);
+        return props;
+    }
+
     /**
      * Poll until {@code count} messages are collected, every assigned partition has reached its
      * end offset, or {@code deadlineNs} expires. Empty polls retry until caught-up or deadline —
@@ -1335,15 +1565,10 @@ public final class KafkaAgent {
     }
 
     private static String tryDecodeUtf8(byte[] bytes) {
-        try {
-            String text = new String(bytes, StandardCharsets.UTF_8);
-            // Verify round-trip
-            byte[] reEncoded = text.getBytes(StandardCharsets.UTF_8);
-            if (Arrays.equals(bytes, reEncoded)) {
-                return text;
-            }
-        } catch (Exception ignored) {}
-        return null;
+        String text = new String(bytes, StandardCharsets.UTF_8);
+        // Replacement characters change the bytes on round-trip, identifying invalid UTF-8 without exceptions.
+        byte[] reEncoded = text.getBytes(StandardCharsets.UTF_8);
+        return Arrays.equals(bytes, reEncoded) ? text : null;
     }
 
     static Long normalizePeekOffset(long requestedOffset, long beginningOffset, long endOffset) {
diff --git a/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java b/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java
index 6ebe51311..c76cfd947 100644
--- a/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java
+++ b/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java
@@ -1,24 +1,282 @@
 package com.dbx.agent.kafka;
 
+import com.google.gson.JsonObject;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.google.gson.JsonParser;
+import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.Config;
+import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 
 class KafkaAgentTest {
+    @TempDir
+    Path tempDir;
+
+    @Test
+    void resolvesBootstrapServersFromKafka11ZooKeeperRegistrationWithChroot() throws Exception {
+        Path snapshots = Files.createDirectory(tempDir.resolve("snapshots"));
+        Path logs = Files.createDirectory(tempDir.resolve("logs"));
+        ZooKeeperServer server = new ZooKeeperServer(snapshots.toFile(), logs.toFile(), 2_000);
+        NIOServerCnxnFactory factory = new NIOServerCnxnFactory();
+        factory.configure(new InetSocketAddress("127.0.0.1", 0), 10);
+        factory.startup(server);
+
+        ZooKeeper client = null;
+        String previousSaslSetting = System.getProperty("zookeeper.sasl.client");
+        try {
+            CountDownLatch connected = new CountDownLatch(1);
+            System.setProperty("zookeeper.sasl.client", "false");
+            client = new ZooKeeper("127.0.0.1:" + factory.getLocalPort(), 5_000, event -> {
+                if (event.getState() == Watcher.Event.KeeperState.SyncConnected) connected.countDown();
+            });
+            assertTrue(connected.await(5, TimeUnit.SECONDS));
+            client.create("/kafka", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+            client.create("/kafka/brokers", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+            client.create("/kafka/brokers/ids", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+            client.create(
+                "/kafka/brokers/ids/0",
+                "{\"listener_security_protocol_map\":{\"PLAINTEXT\":\"PLAINTEXT\"},\"endpoints\":[\"PLAINTEXT://legacy-broker:9092\"]}".getBytes(StandardCharsets.UTF_8),
+                ZooDefs.Ids.OPEN_ACL_UNSAFE,
+                CreateMode.EPHEMERAL
+            );
+
+            JsonObject connection = new JsonObject();
+            connection.addProperty("zookeeper_connect_string", "127.0.0.1:" + factory.getLocalPort() + "/kafka");
+            connection.addProperty("security_protocol", "PLAINTEXT");
+            connection.addProperty("zookeeper_connection_timeout_ms", 5_000);
+
+            JsonObject resolved = KafkaAgent.resolveBrokerConnection(connection);
+
+            assertEquals("legacy-broker:9092", resolved.get("bootstrap_servers").getAsString());
+        } finally {
+            if (client != null) client.close();
+            factory.shutdown();
+            server.shutdown();
+            server.getTxnLogFactory().close();
+            if (previousSaslSetting == null) {
+                System.clearProperty("zookeeper.sasl.client");
+            } else {
+                System.setProperty("zookeeper.sasl.client", previousSaslSetting);
+            }
+        }
+    }
+
+    @Test
+    void zooKeeperClientConfigPreservesSaslAndTlsSystemDefaults() {
+        Map<String, String> previous = preserveSystemProperties(
+            "zookeeper.sasl.client",
+            "zookeeper.sasl.clientconfig",
+            "zookeeper.client.secure",
+            "zookeeper.clientCnxnSocket",
+            "zookeeper.ssl.trustStore.location",
+            "java.security.auth.login.config"
+        );
+        try {
+            System.setProperty("zookeeper.sasl.client", "true");
+            System.setProperty("zookeeper.sasl.clientconfig", "DbxZooKeeperClient");
+            System.setProperty("zookeeper.client.secure", "true");
+            System.setProperty("zookeeper.clientCnxnSocket", "org.apache.zookeeper.ClientCnxnSocketNetty");
+            System.setProperty("zookeeper.ssl.trustStore.location", "/etc/dbx/zookeeper-truststore.p12");
+            System.setProperty("java.security.auth.login.config", "/etc/dbx/zookeeper-jaas.conf");
+
+            ZKClientConfig config = KafkaAgent.zooKeeperClientConfig(new JsonObject());
+
+            assertTrue(config.isSaslClientEnabled());
+            assertEquals("DbxZooKeeperClient", config.getProperty("zookeeper.sasl.clientconfig"));
+            assertEquals("true", config.getProperty("zookeeper.client.secure"));
+            assertEquals(
+                "org.apache.zookeeper.ClientCnxnSocketNetty",
+                config.getProperty("zookeeper.clientCnxnSocket")
+            );
+            assertEquals(
+                "/etc/dbx/zookeeper-truststore.p12",
+                config.getProperty("zookeeper.ssl.trustStore.location")
+            );
+            assertEquals("/etc/dbx/zookeeper-jaas.conf", config.getJaasConfKey());
+        } finally {
+            restoreSystemProperties(previous);
+        }
+    }
+
+    @Test
+    void zooKeeperClientConfigAppliesPerConnectionSaslAndTlsOverridesWithoutChangingJvmState() {
+        Map<String, String> previous = preserveSystemProperties(
+            "zookeeper.sasl.client",
+            "zookeeper.sasl.clientconfig",
+            "zookeeper.client.secure",
+            "zookeeper.clientCnxnSocket",
+            "zookeeper.ssl.keyStore.location"
+        );
+        try {
+            System.setProperty("zookeeper.sasl.client", "false");
+            System.setProperty("zookeeper.client.secure", "false");
+
+            JsonObject properties = new JsonObject();
+            properties.addProperty("zookeeper.sasl.client", "true");
+            properties.addProperty("zookeeper.sasl.clientconfig", "DbxZooKeeperClient");
+            properties.addProperty("zookeeper.client.secure", "true");
+            properties.addProperty("zookeeper.clientCnxnSocket", "org.apache.zookeeper.ClientCnxnSocketNetty");
+            properties.addProperty("zookeeper.ssl.keyStore.location", "/etc/dbx/zookeeper-keystore.p12");
+            properties.addProperty("security.protocol", "SASL_SSL");
+            JsonObject connection = new JsonObject();
+            connection.add("properties", properties);
+
+            ZKClientConfig config = KafkaAgent.zooKeeperClientConfig(connection);
+
+            assertTrue(config.isSaslClientEnabled());
+            assertEquals("DbxZooKeeperClient", config.getProperty("zookeeper.sasl.clientconfig"));
+            assertEquals("true", config.getProperty("zookeeper.client.secure"));
+            assertEquals(
+                "org.apache.zookeeper.ClientCnxnSocketNetty",
+                config.getProperty("zookeeper.clientCnxnSocket")
+            );
+            assertEquals(
+                "/etc/dbx/zookeeper-keystore.p12",
+                config.getProperty("zookeeper.ssl.keyStore.location")
+            );
+            assertNull(config.getProperty("security.protocol"));
+            assertEquals("false", System.getProperty("zookeeper.sasl.client"));
+            assertEquals("false", System.getProperty("zookeeper.client.secure"));
+        } finally {
+            restoreSystemProperties(previous);
+        }
+    }
+
+    @Test
+    void brokerEndpointsUseListenerSecurityProtocolMapForNamedListenersAndKeepBrokerOrder() {
+        List<JsonObject> registrations = Arrays.asList(
+            broker("{\"listener_security_protocol_map\":{\"INTERNAL\":\"PLAINTEXT\",\"CLIENT\":\"SASL_SSL\"},\"endpoints\":[\"INTERNAL://broker-2:9092\",\"CLIENT://public-2:9093\"]}"),
+            broker("{\"listener_security_protocol_map\":{\"INTERNAL\":\"PLAINTEXT\",\"CLIENT\":\"SASL_SSL\"},\"endpoints\":[\"CLIENT://public-1:9093\",\"INTERNAL://broker-1:9092\"]}")
+        );
+
+        assertEquals("public-2:9093,public-1:9093", KafkaAgent.brokerEndpoints(registrations, "SASL_SSL"));
+    }
+
+    @Test
+    void kafkaClientPropertiesExcludeZooKeeperSecuritySettings() {
+        JsonObject properties = new JsonObject();
+        properties.addProperty("client.id", "dbx");
+        properties.addProperty("zookeeper.sasl.client", "true");
+        properties.addProperty("zookeeper.ssl.trustStore.password", "secret");
+        JsonObject connection = new JsonObject();
+        connection.add("properties", properties);
+
+        Properties kafkaProperties = new Properties();
+        KafkaAgent.applyConnectionProperties(connection, kafkaProperties);
+
+        assertEquals("dbx", kafkaProperties.getProperty("client.id"));
+        assertNull(kafkaProperties.getProperty("zookeeper.sasl.client"));
+        assertNull(kafkaProperties.getProperty("zookeeper.ssl.trustStore.password"));
+    }
+
+    @Test
+    void brokerEndpointsFallBackToLegacyHostAndPort() {
+        assertEquals("legacy-broker:9092", KafkaAgent.brokerEndpoints(
+            Collections.singletonList(broker("{\"host\":\"legacy-broker\",\"port\":9092}")), "PLAINTEXT"));
+    }
+
+    @Test
+    void brokerEndpointsSkipMalformedRegistrationWhenAnotherBrokerIsUsable() {
+        assertEquals("healthy-broker:9092", KafkaAgent.brokerEndpoints(Arrays.asList(
+            broker("{\"host\":\"broken\",\"port\":\"not-a-port\"}"),
+            broker("{\"host\":\"healthy-broker\",\"port\":9092}")
+        ), "PLAINTEXT"));
+    }
+
+    @Test
+    void brokerEndpointsRejectRegistrationsWithoutUsableAddresses() {
+        var error = org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException.class,
+            () -> KafkaAgent.brokerEndpoints(Collections.singletonList(broker("{\"endpoints\":[]}")), "PLAINTEXT"));
+        assertTrue(error.getMessage().contains("usable Kafka broker endpoints"));
+    }
+
+    @Test
+    void peekConsumerPropertiesReuseResolvedConnection() {
+        JsonObject resolved = new JsonObject();
+        resolved.addProperty("bootstrap_servers", "legacy-broker:9092");
+        resolved.addProperty("security_protocol", "PLAINTEXT");
+
+        Properties properties = KafkaAgent.peekConsumerProperties(resolved, 25);
+
+        assertEquals("legacy-broker:9092", properties.getProperty("bootstrap.servers"));
+        assertEquals(25, properties.get("max.poll.records"));
+    }
+
+    @Test
+    void aclDisabledDetectionOnlyAcceptsKnownAuthorizerErrors() {
+        Exception disabled = new RuntimeException(
+            "ACL probe failed",
+            new IllegalStateException("No Authorizer is configured on the broker")
+        );
+
+        assertTrue(KafkaAgent.isAclDisabledError(disabled));
+        assertFalse(KafkaAgent.isAclDisabledError(new RuntimeException("Timed out waiting for broker response")));
+    }
+
+    @Test
+    void legacyTopicConfigAppliesSetAndDeleteWithoutLosingExistingOverrides() {
+        Config current = new Config(Arrays.asList(
+            new ConfigEntry("cleanup.policy", "delete"),
+            new ConfigEntry("retention.ms", "60000"),
+            new ConfigEntry(
+                "segment.bytes",
+                "1073741824",
+                ConfigEntry.ConfigSource.DYNAMIC_BROKER_CONFIG,
+                false,
+                false,
+                Collections.emptyList(),
+                ConfigEntry.ConfigType.LONG,
+                null
+            )
+        ));
+        List<AlterConfigOp> ops = Arrays.asList(
+            new AlterConfigOp(new ConfigEntry("retention.ms", "120000"), AlterConfigOp.OpType.SET),
+            new AlterConfigOp(new ConfigEntry("cleanup.policy", null), AlterConfigOp.OpType.DELETE)
+        );
+
+        Map<String, String> merged = KafkaAgent.legacyTopicConfig(current, ops);
+
+        assertEquals(Collections.singletonMap("retention.ms", "120000"), merged);
+    }
+
+    @Test
+    void legacyTopicConfigRejectsAppendAndSubtractOperations() {
+        Config current = new Config(Collections.singletonList(new ConfigEntry("cleanup.policy", "delete")));
+        AlterConfigOp append = new AlterConfigOp(new ConfigEntry("cleanup.policy", "compact"), AlterConfigOp.OpType.APPEND);
+
+        var error = org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException.class,
+            () -> KafkaAgent.legacyTopicConfig(current, Collections.singletonList(append)));
+        assertTrue(error.getMessage().contains("APPEND"));
+    }
     @Test
     void normalizesPeekOffsetToEarliestAvailableOffset() {
         assertEquals(5L, KafkaAgent.normalizePeekOffset(0, 5, 10));
@@ -241,4 +499,24 @@ class KafkaAgentTest {
             }
         }
     }
+
+    private static JsonObject broker(String json) {
+        return JsonParser.parseString(json).getAsJsonObject();
+    }
+
+    private static Map<String, String> preserveSystemProperties(String... keys) {
+        Map<String, String> previous = new HashMap<>();
+        for (String key : keys) previous.put(key, System.getProperty(key));
+        return previous;
+    }
+
+    private static void restoreSystemProperties(Map<String, String> properties) {
+        for (Map.Entry<String, String> entry : properties.entrySet()) {
+            if (entry.getValue() == null) {
+                System.clearProperty(entry.getKey());
+            } else {
+                System.setProperty(entry.getKey(), entry.getValue());
+            }
+        }
+    }
 }
diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue
index 477154324..180cbcc5e 100644
--- a/apps/desktop/src/components/connection/ConnectionDialog.vue
+++ b/apps/desktop/src/components/connection/ConnectionDialog.vue
@@ -46,7 +46,7 @@ import { SQLITE_DATABASE_FILE_EXTENSIONS } from "@/lib/database/databaseFileDete
 import { connectionAttemptOriginalErrorMessage, connectionAttemptTimeoutMessage, connectionAttemptTimeoutMs } from "@/lib/connection/connectionAttemptTimeout";
 import { appendConnectionErrorHints, isJdbcMissingRuntimeDependencyError } from "@/lib/connection/connectionErrorHints";
 import { postgresTlsModeForForm } from "@/lib/connection/postgresTlsMode";
-import { normalizeKafkaBootstrapServers } from "@/lib/connection/kafkaBootstrapServers";
+import { buildMqKafkaConnectionExtra, mqKafkaConnectionTarget, resolveMqKafkaConnectionSource, type MqKafkaConnectionSource } from "@/lib/connection/mqKafkaConnection";
 import { assertCompleteDatabaseCategories, databaseSelectionForCategory } from "@/lib/connection/databaseCategoryOptions";
 import { normalizeRocketmqNamesrvAddr } from "@/lib/connection/rocketmqNamesrv";
 import { normalizeRabbitmqAddresses } from "@/lib/connection/rabbitmqAddresses";
@@ -517,11 +517,13 @@ const configTab = ref<ConfigTab>("connection");
 const MQ_KAFKA_SECURITY_PROTOCOL_AUTO = "__auto";
 const mqAdminUrl = ref("http://127.0.0.1:8080");
 const mqSystemKind = ref<MqSystemKind>("pulsar");
+const mqKafkaConnectionSource = ref<MqKafkaConnectionSource>("bootstrap");
 const mqRocketmqNamesrvAddr = ref("127.0.0.1:9876");
 const mqRocketmqClusterName = ref("");
 const mqRabbitmqAddresses = ref("127.0.0.1:5672");
 const mqRabbitmqVirtualHost = ref("/");
 const mqKafkaBootstrapServers = ref("127.0.0.1:9092");
+const mqKafkaZooKeeperServers = ref("");
 const mqKafkaSecurityProtocol = ref(MQ_KAFKA_SECURITY_PROTOCOL_AUTO);
 const mqKafkaSaslMechanism = ref("PLAIN");
 const mqKafkaKerberosPrincipal = ref("");
@@ -575,6 +577,10 @@ const mqKafkaSecurityProtocolOptions = computed(() => [
   { value: "SASL_PLAINTEXT", label: "SASL_PLAINTEXT" },
   { value: "SASL_SSL", label: "SASL_SSL" },
 ]);
+const mqKafkaConnectionSourceOptions = computed(() => [
+  { value: "bootstrap" as const, label: t("connection.mqKafkaConnectionSourceBootstrap") },
+  { value: "zookeeper" as const, label: t("connection.mqKafkaConnectionSourceZooKeeper") },
+]);
 const mqKafkaSaslMechanismOptions = [
   { value: "PLAIN", label: "PLAIN" },
   { value: "SCRAM-SHA-256", label: "SCRAM-SHA-256" },
@@ -892,7 +898,9 @@ function resetMqFields(config?: Partial<MqAdminConfig>) {
   mqSystemKind.value = systemKind;
   const storedAdminUrl = config?.adminUrl?.trim() || (config ? mqExtraString(config as Record<string, unknown>, "admin_url").trim() : "");
   mqAdminUrl.value = storedAdminUrl || (systemKind === "kafka" || systemKind === "rocketmq" || systemKind === "rabbitmq" ? "" : "http://127.0.0.1:8080");
+  mqKafkaConnectionSource.value = resolveMqKafkaConnectionSource(extra);
   mqKafkaBootstrapServers.value = mqExtraString(extra, "bootstrapServers") || "127.0.0.1:9092";
+  mqKafkaZooKeeperServers.value = mqExtraString(extra, "zookeeperServers");
   mqRocketmqNamesrvAddr.value = mqExtraString(extra, "namesrvAddr") || mqExtraString(extra, "namesrv_addr") || "127.0.0.1:9876";
   mqRocketmqClusterName.value = mqExtraString(extra, "clusterName") || mqExtraString(extra, "cluster_name");
   mqRabbitmqAddresses.value = mqExtraString(extra, "addresses") || "127.0.0.1:5672";
@@ -969,7 +977,7 @@ watch(selectedType, () => {
 
 watch(mqSystemKind, (kind) => {
   if (kind === "kafka") {
-    if (!mqKafkaBootstrapServers.value.trim()) mqKafkaBootstrapServers.value = "127.0.0.1:9092";
+    if (mqKafkaConnectionSource.value === "bootstrap" && !mqKafkaBootstrapServers.value.trim()) mqKafkaBootstrapServers.value = "127.0.0.1:9092";
     if (!isMqAuthKindAllowedForSystem(kind, mqAuthKind.value)) mqAuthKind.value = "none";
     return;
   }
@@ -1108,9 +1116,14 @@ function buildMqTokenSigning() {
 function buildMqAdminConfig(): MqAdminConfig {
   const systemKind = mqSystemKind.value;
   if (systemKind === "kafka") {
-    const bootstrapServers = normalizeKafkaBootstrapServers(mqKafkaBootstrapServers.value);
-    const extra: Record<string, unknown> = { bootstrapServers };
-    const securityProtocol = mqKafkaSecurityProtocol.value === MQ_KAFKA_SECURITY_PROTOCOL_AUTO ? "" : mqKafkaSecurityProtocol.value.trim();
+    const configuredSecurityProtocol = mqKafkaSecurityProtocol.value === MQ_KAFKA_SECURITY_PROTOCOL_AUTO ? "" : mqKafkaSecurityProtocol.value;
+    const extra: Record<string, unknown> = buildMqKafkaConnectionExtra({
+      connectionSource: mqKafkaConnectionSource.value,
+      bootstrapServers: mqKafkaBootstrapServers.value,
+      zookeeperServers: mqKafkaZooKeeperServers.value,
+      securityProtocol: configuredSecurityProtocol,
+    });
+    const securityProtocol = mqExtraString(extra, "securityProtocol");
     const saslMechanism = mqAuthKind.value === "kerberos" ? "GSSAPI" : mqKafkaSaslMechanism.value.trim();
     const properties: Record<string, string> = {};
     if (securityProtocol) extra.securityProtocol = securityProtocol;
@@ -1518,18 +1531,17 @@ function applyMqAdminUrl(config: LegacyConnectionConfig, adminUrl: string) {
   config.ssl = parsed.protocol === "https:";
 }
 
-function applyMqKafkaBootstrapServers(config: LegacyConnectionConfig, bootstrapServers: string, securityProtocol?: string) {
-  const first = normalizeKafkaBootstrapServers(bootstrapServers).split(",")[0];
-  if (!first) throw new Error(t("connection.mqBootstrapServersRequired"));
-  let parsed: URL;
-  try {
-    parsed = new URL(`kafka://${first}`);
-  } catch {
-    throw new Error(t("connection.mqBootstrapServersInvalid"));
-  }
-  config.host = parsed.hostname;
-  config.port = Number(parsed.port) || 9092;
-  config.ssl = securityProtocol === "SSL" || securityProtocol === "SASL_SSL";
+function applyMqKafkaConnectionTarget(config: LegacyConnectionConfig, extra: Record<string, unknown>) {
+  const source = resolveMqKafkaConnectionSource(extra);
+  const target = mqKafkaConnectionTarget({
+    connectionSource: source,
+    bootstrapServers: mqExtraString(extra, "bootstrapServers"),
+    zookeeperServers: mqExtraString(extra, "zookeeperServers"),
+    securityProtocol: mqExtraString(extra, "securityProtocol"),
+  });
+  config.host = target.host;
+  config.port = target.port;
+  config.ssl = target.ssl;
 }
 
 function applyMqRabbitmqAddresses(config: LegacyConnectionConfig, addresses: string) {
@@ -2548,7 +2560,7 @@ const connectionLabelTopClass = `${connectionLabelClass} mt-2`;
 const connectionLabelSmallPaddedClass = `${connectionLabelClass} pt-2 text-xs`;
 const hasRequiredConnectionTarget = computed(() => {
   if (form.value.db_type === "mq") {
-    if (mqSystemKind.value === "kafka") return !!mqKafkaBootstrapServers.value.trim();
+    if (mqSystemKind.value === "kafka") return mqKafkaConnectionSource.value === "zookeeper" ? !!mqKafkaZooKeeperServers.value.trim() : !!mqKafkaBootstrapServers.value.trim();
     if (mqSystemKind.value === "rocketmq") return !!mqRocketmqNamesrvAddr.value.trim();
     if (mqSystemKind.value === "rabbitmq") return !!mqRabbitmqAddresses.value.trim();
     return !!mqAdminUrl.value.trim();
@@ -2881,7 +2893,7 @@ function connectionConfigForSubmit(id: string, generatedName = ""): ConnectionCo
     config.driver_label = MQ_DRIVER_LABELS[mqConfig.systemKind];
     if (mqConfig.systemKind === "kafka") {
       const extra = mqExtraRecord(mqConfig);
-      applyMqKafkaBootstrapServers(config, mqExtraString(extra, "bootstrapServers"), mqExtraString(extra, "securityProtocol"));
+      applyMqKafkaConnectionTarget(config, extra);
     } else if (mqConfig.systemKind === "rocketmq") {
       const extra = mqExtraRecord(mqConfig);
       applyMqRocketmqNamesrv(config, mqExtraString(extra, "namesrvAddr") || mqExtraString(extra, "namesrv_addr"));
@@ -4769,9 +4781,26 @@ function openExternalUrl(url: string) {
                 <template v-else-if="form.db_type === 'mq'">
                   <template v-if="mqSystemKind === 'kafka'">
                     <div class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.mqKafkaConnectionSource") }}</Label>
+                      <Select v-model="mqKafkaConnectionSource">
+                        <SelectTrigger class="col-span-3 h-9">
+                          <SelectValue />
+                        </SelectTrigger>
+                        <SelectContent>
+                          <SelectItem v-for="option in mqKafkaConnectionSourceOptions" :key="option.value" :value="option.value">
+                            {{ option.label }}
+                          </SelectItem>
+                        </SelectContent>
+                      </Select>
+                    </div>
+                    <div v-if="mqKafkaConnectionSource === 'bootstrap'" class="grid grid-cols-4 items-center gap-4">
                       <Label :class="connectionLabelClass">{{ t("connection.mqBootstrapServers") }}</Label>
                       <Input v-model="mqKafkaBootstrapServers" class="col-span-3" :placeholder="t('connection.mqBootstrapServersPlaceholder')" />
                     </div>
+                    <div v-else class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.mqKafkaZooKeeperServers") }}</Label>
+                      <Input v-model="mqKafkaZooKeeperServers" class="col-span-3" :placeholder="t('connection.mqKafkaZooKeeperServersPlaceholder')" />
+                    </div>
                     <div class="grid grid-cols-4 items-center gap-4">
                       <Label :class="connectionLabelClass">{{ t("connection.mqSecurity") }}</Label>
                       <Select v-model="mqKafkaSecurityProtocol">
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index a4662f712..0e4704c97 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -356,6 +356,11 @@ export default {
     mqBootstrapServersPlaceholder: "127.0.0.1:9092",
     mqBootstrapServersRequired: "Kafka bootstrap servers are required",
     mqBootstrapServersInvalid: "Kafka bootstrap servers are invalid",
+    mqKafkaConnectionSource: "Connection Source",
+    mqKafkaConnectionSourceBootstrap: "Bootstrap Servers",
+    mqKafkaConnectionSourceZooKeeper: "ZooKeeper (Kafka 1.x)",
+    mqKafkaZooKeeperServers: "ZooKeeper Servers",
+    mqKafkaZooKeeperServersPlaceholder: "zk1:2181,zk2:2181/kafka",
     mqSystemRabbitMq: "RabbitMQ",
     mqRabbitmqAddresses: "Addresses",
     mqRabbitmqAddressesPlaceholder: "127.0.0.1:5672",
diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts
index 4046c2cc1..84dc2cfee 100644
--- a/apps/desktop/src/i18n/locales/es.ts
+++ b/apps/desktop/src/i18n/locales/es.ts
@@ -487,6 +487,11 @@ export default withEnglishFallback({
     mqBootstrapServersPlaceholder: "127.0.0.1:9092",
     mqBootstrapServersRequired: "Los bootstrap servers de Kafka son obligatorios",
     mqBootstrapServersInvalid: "Los bootstrap servers de Kafka no son válidos",
+    mqKafkaConnectionSource: "Origen de conexión",
+    mqKafkaConnectionSourceBootstrap: "Bootstrap Servers",
+    mqKafkaConnectionSourceZooKeeper: "ZooKeeper (Kafka 1.x)",
+    mqKafkaZooKeeperServers: "Servidores ZooKeeper",
+    mqKafkaZooKeeperServersPlaceholder: "zk1:2181,zk2:2181/kafka",
     mqSystemRabbitMq: "RabbitMQ",
     mqRabbitmqAddresses: "Direcciones",
     mqRabbitmqAddressesPlaceholder: "127.0.0.1:5672",
diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts
index f34f2b77b..dc4ba774e 100644
--- a/apps/desktop/src/i18n/locales/it.ts
+++ b/apps/desktop/src/i18n/locales/it.ts
@@ -485,6 +485,11 @@ export default withEnglishFallback({
     mqBootstrapServersPlaceholder: "127.0.0.1:9092",
     mqBootstrapServersRequired: "I bootstrap server Kafka sono obbligatori",
     mqBootstrapServersInvalid: "I bootstrap server Kafka non sono validi",
+    mqKafkaConnectionSource: "Origine connessione",
+    mqKafkaConnectionSourceBootstrap: "Bootstrap Servers",
+    mqKafkaConnectionSourceZooKeeper: "ZooKeeper (Kafka 1.x)",
+    mqKafkaZooKeeperServers: "Server ZooKeeper",
+    mqKafkaZooKeeperServersPlaceholder: "zk1:2181,zk2:2181/kafka",
     mqSystemRabbitMq: "RabbitMQ",
     mqRabbitmqAddresses: "Indirizzi",
     mqRabbitmqAddressesPlaceholder: "127.0.0.1:5672",
diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts
index 98a4fe597..70624bc0a 100644
--- a/apps/desktop/src/i18n/locales/ja.ts
+++ b/apps/desktop/src/i18n/locales/ja.ts
@@ -485,6 +485,11 @@ export default withEnglishFallback({
     mqBootstrapServersPlaceholder: "127.0.0.1:9092",
     mqBootstrapServersRequired: "Kafka Bootstrap Servers は必須です",
     mqBootstrapServersInvalid: "Kafka Bootstrap Servers が無効です",
+    mqKafkaConnectionSource: "接続元",
+    mqKafkaConnectionSourceBootstrap: "Bootstrap Servers",
+    mqKafkaConnectionSourceZooKeeper: "ZooKeeper (Kafka 1.x)",
+    mqKafkaZooKeeperServers: "ZooKeeper サーバー",
+    mqKafkaZooKeeperServersPlaceholder: "zk1:2181,zk2:2181/kafka",
     mqSystemRabbitMq: "RabbitMQ",
     mqRabbitmqAddresses: "アドレス",
     mqRabbitmqAddressesPlaceholder: "127.0.0.1:5672",
diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts
index af5f318fe..acabe648b 100644
--- a/apps/desktop/src/i18n/locales/pt-BR.ts
+++ b/apps/desktop/src/i18n/locales/pt-BR.ts
@@ -486,6 +486,11 @@ export default withEnglishFallback({
     mqBootstrapServersPlaceholder: "127.0.0.1:9092",
     mqBootstrapServersRequired: "Bootstrap Servers do Kafka são obrigatórios",
     mqBootstrapServersInvalid: "Bootstrap Servers do Kafka são inválidos",
+    mqKafkaConnectionSource: "Origem da conexão",
+    mqKafkaConnectionSourceBootstrap: "Bootstrap Servers",
+    mqKafkaConnectionSourceZooKeeper: "ZooKeeper (Kafka 1.x)",
+    mqKafkaZooKeeperServers: "Servidores ZooKeeper",
+    mqKafkaZooKeeperServersPlaceholder: "zk1:2181,zk2:2181/kafka",
     mqSystemRabbitMq: "RabbitMQ",
     mqRabbitmqAddresses: "Endereços",
     mqRabbitmqAddressesPlaceholder: "127.0.0.1:5672",
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index 00f017447..7fbe7e85c 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -358,6 +358,11 @@ export default withEnglishFallback({
     mqBootstrapServersPlaceholder: "127.0.0.1:9092",
     mqBootstrapServersRequired: "Kafka Bootstrap Servers 不能为空",
     mqBootstrapServersInvalid: "Kafka Bootstrap Servers 无效",
+    mqKafkaConnectionSource: "连接来源",
+    mqKafkaConnectionSourceBootstrap: "Bootstrap Servers",
+    mqKafkaConnectionSourceZooKeeper: "ZooKeeper（Kafka 1.x）",
+    mqKafkaZooKeeperServers: "ZooKeeper Servers",
+    mqKafkaZooKeeperServersPlaceholder: "zk1:2181,zk2:2181/kafka",
     mqSystemRabbitMq: "RabbitMQ",
     mqRabbitmqAddresses: "地址",
     mqRabbitmqAddressesPlaceholder: "127.0.0.1:5672",
diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts
index e69a26658..9caadf126 100644
--- a/apps/desktop/src/i18n/locales/zh-TW.ts
+++ b/apps/desktop/src/i18n/locales/zh-TW.ts
@@ -486,6 +486,11 @@ export default withEnglishFallback({
     mqBootstrapServersPlaceholder: "127.0.0.1:9092",
     mqBootstrapServersRequired: "Kafka Bootstrap Servers 不能為空",
     mqBootstrapServersInvalid: "Kafka Bootstrap Servers 無效",
+    mqKafkaConnectionSource: "連線來源",
+    mqKafkaConnectionSourceBootstrap: "Bootstrap Servers",
+    mqKafkaConnectionSourceZooKeeper: "ZooKeeper（Kafka 1.x）",
+    mqKafkaZooKeeperServers: "ZooKeeper Servers",
+    mqKafkaZooKeeperServersPlaceholder: "zk1:2181,zk2:2181/kafka",
     mqSystemRabbitMq: "RabbitMQ",
     mqRabbitmqAddresses: "位址",
     mqRabbitmqAddressesPlaceholder: "127.0.0.1:5672",
diff --git a/apps/desktop/src/lib/__tests__/connection/kafkaBootstrapServers.spec.ts b/apps/desktop/src/lib/__tests__/connection/kafkaBootstrapServers.spec.ts
index f102d3276..e8bff2e82 100644
--- a/apps/desktop/src/lib/__tests__/connection/kafkaBootstrapServers.spec.ts
+++ b/apps/desktop/src/lib/__tests__/connection/kafkaBootstrapServers.spec.ts
@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { normalizeKafkaBootstrapServers } from "@/lib/connection/kafkaBootstrapServers";
+import { normalizeKafkaBootstrapServers, parseKafkaBootstrapServers, resolveKafkaSecurityProtocol } from "@/lib/connection/kafkaBootstrapServers";
 
 describe("Kafka bootstrap servers", () => {
   it("keeps comma-separated bootstrap servers", () => {
@@ -14,11 +14,34 @@ describe("Kafka bootstrap servers", () => {
     expect(normalizeKafkaBootstrapServers("[::1]:9092;[2001:db8::1]:9092")).toBe("[::1]:9092,[2001:db8::1]:9092");
   });
 
-  it("rejects bootstrap servers with URL schemes", () => {
-    expect(() => normalizeKafkaBootstrapServers("PLAINTEXT://broker1:9092,broker2:9092")).toThrow("Kafka bootstrap servers must be host:port values without a URL scheme");
+  it("normalizes Kafka listener URIs and exposes the inferred security protocol", () => {
+    expect(parseKafkaBootstrapServers("PLAINTEXT://broker1:9092, broker2:9092")).toEqual({
+      bootstrapServers: "broker1:9092,broker2:9092",
+      inferredSecurityProtocol: "PLAINTEXT",
+    });
+    expect(parseKafkaBootstrapServers("SASL_SSL://secure-broker:9093")).toEqual({
+      bootstrapServers: "secure-broker:9093",
+      inferredSecurityProtocol: "SASL_SSL",
+    });
+  });
+
+  it("rejects bootstrap servers that declare conflicting security protocols", () => {
+    expect(() => normalizeKafkaBootstrapServers("SSL://broker1:9093,PLAINTEXT://broker2:9092")).toThrow("Kafka bootstrap servers must use one security protocol");
+  });
+
+  it("rejects unknown listener URI schemes", () => {
+    expect(() => normalizeKafkaBootstrapServers("INTERNAL://broker1:9092")).toThrow("Kafka bootstrap server protocol is invalid");
+  });
+
+  it("uses an inferred protocol only when security remains automatic", () => {
+    expect(resolveKafkaSecurityProtocol("", "SASL_SSL")).toBe("SASL_SSL");
+    expect(resolveKafkaSecurityProtocol("SSL", "SASL_SSL")).toBe("SSL");
+    expect(resolveKafkaSecurityProtocol("", undefined)).toBe("");
   });
 
   it("rejects invalid bootstrap server values", () => {
     expect(() => normalizeKafkaBootstrapServers("broker1:9092/path,broker2:9092")).toThrow("Kafka bootstrap servers are invalid");
+    expect(() => normalizeKafkaBootstrapServers("broker1")).toThrow("Kafka bootstrap servers must be host:port values");
+    expect(() => normalizeKafkaBootstrapServers("broker1:70000")).toThrow("Kafka bootstrap servers are invalid");
   });
 });
diff --git a/apps/desktop/src/lib/__tests__/connection/mqKafkaConnection.spec.ts b/apps/desktop/src/lib/__tests__/connection/mqKafkaConnection.spec.ts
new file mode 100644
index 000000000..3e6d366bb
--- /dev/null
+++ b/apps/desktop/src/lib/__tests__/connection/mqKafkaConnection.spec.ts
@@ -0,0 +1,76 @@
+import { describe, expect, it } from "vitest";
+import { buildMqKafkaConnectionExtra, mqKafkaConnectionTarget, normalizeMqKafkaZooKeeperServers, resolveMqKafkaConnectionSource } from "@/lib/connection/mqKafkaConnection";
+
+describe("MQ Kafka connection", () => {
+  it("keeps existing configurations on the Bootstrap source", () => {
+    expect(resolveMqKafkaConnectionSource({ bootstrapServers: "broker-1:9092" })).toBe("bootstrap");
+    expect(resolveMqKafkaConnectionSource({})).toBe("bootstrap");
+  });
+
+  it("recognizes explicit and legacy ZooKeeper discovery configurations", () => {
+    expect(resolveMqKafkaConnectionSource({ connectionSource: "zookeeper", zookeeperServers: "zk-1:2181" })).toBe("zookeeper");
+    expect(resolveMqKafkaConnectionSource({ zookeeperServers: "zk-legacy:2181" })).toBe("zookeeper");
+  });
+
+  it("builds Bootstrap extra fields from listener URIs without persisting the scheme", () => {
+    expect(
+      buildMqKafkaConnectionExtra({
+        connectionSource: "bootstrap",
+        bootstrapServers: "SASL_SSL://broker-1:9093, broker-2:9093",
+        zookeeperServers: "ignored:2181",
+        securityProtocol: "",
+      }),
+    ).toEqual({ bootstrapServers: "broker-1:9093,broker-2:9093", securityProtocol: "SASL_SSL" });
+  });
+
+  it("keeps an explicit security protocol ahead of the listener URI hint", () => {
+    expect(
+      buildMqKafkaConnectionExtra({
+        connectionSource: "bootstrap",
+        bootstrapServers: "PLAINTEXT://broker-1:9092",
+        zookeeperServers: "",
+        securityProtocol: "SSL",
+      }),
+    ).toEqual({ bootstrapServers: "broker-1:9092", securityProtocol: "SSL" });
+  });
+
+  it("builds ZooKeeper discovery fields without a fake Bootstrap address", () => {
+    expect(
+      buildMqKafkaConnectionExtra({
+        connectionSource: "zookeeper",
+        bootstrapServers: "ignored:9092",
+        zookeeperServers: "zookeeper://zk-1:2181; zk-2:2181/kafka",
+        securityProtocol: "PLAINTEXT",
+      }),
+    ).toEqual({
+      connectionSource: "zookeeper",
+      zookeeperServers: "zk-1:2181,zk-2:2181/kafka",
+      securityProtocol: "PLAINTEXT",
+    });
+  });
+
+  it("rejects malformed ZooKeeper discovery addresses", () => {
+    expect(() => normalizeMqKafkaZooKeeperServers("zk-1:not-a-port/kafka")).toThrow("Kafka ZooKeeper servers are invalid");
+    expect(() => normalizeMqKafkaZooKeeperServers("zk-1:70000/kafka")).toThrow("Kafka ZooKeeper servers are invalid");
+  });
+
+  it("maps the active source to the generic connection target", () => {
+    expect(
+      mqKafkaConnectionTarget({
+        connectionSource: "bootstrap",
+        bootstrapServers: "SSL://broker-1:9093,broker-2:9093",
+        zookeeperServers: "",
+        securityProtocol: "",
+      }),
+    ).toEqual({ host: "broker-1", port: 9093, ssl: true });
+
+    expect(
+      mqKafkaConnectionTarget({
+        connectionSource: "zookeeper",
+        bootstrapServers: "",
+        zookeeperServers: "zk-1:2281,zk-2:2181/kafka",
+        securityProtocol: "SASL_SSL",
+      }),
+    ).toEqual({ host: "zk-1", port: 2281, ssl: false });
+  });
+});
diff --git a/apps/desktop/src/lib/connection/kafkaBootstrapServers.ts b/apps/desktop/src/lib/connection/kafkaBootstrapServers.ts
index 1b836aff7..11bb2bd8a 100644
--- a/apps/desktop/src/lib/connection/kafkaBootstrapServers.ts
+++ b/apps/desktop/src/lib/connection/kafkaBootstrapServers.ts
@@ -1,4 +1,13 @@
 const KAFKA_BOOTSTRAP_SERVER_SEPARATOR = /[\s,;，；]+/u;
+const KAFKA_BOOTSTRAP_SERVER_SCHEME = /^([a-z][a-z0-9_-]*):\/\/(.+)$/iu;
+const KAFKA_SECURITY_PROTOCOLS = new Set(["PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"] as const);
+
+export type KafkaSecurityProtocol = "PLAINTEXT" | "SSL" | "SASL_PLAINTEXT" | "SASL_SSL";
+
+export interface ParsedKafkaBootstrapServers {
+  bootstrapServers: string;
+  inferredSecurityProtocol?: KafkaSecurityProtocol;
+}
 
 function requireKafkaBootstrapServers(value: string): string {
   const trimmed = value.trim();
@@ -6,28 +15,60 @@ function requireKafkaBootstrapServers(value: string): string {
   return trimmed;
 }
 
-function normalizeKafkaBootstrapServer(server: string): string {
-  if (server.includes("://")) {
-    throw new Error("Kafka bootstrap servers must be host:port values without a URL scheme");
+function normalizeKafkaBootstrapServer(server: string): { address: string; securityProtocol?: KafkaSecurityProtocol } {
+  const schemeMatch = server.match(KAFKA_BOOTSTRAP_SERVER_SCHEME);
+  let address = server;
+  let securityProtocol: KafkaSecurityProtocol | undefined;
+  if (schemeMatch) {
+    const protocol = schemeMatch[1].toUpperCase();
+    if (!KAFKA_SECURITY_PROTOCOLS.has(protocol as KafkaSecurityProtocol)) {
+      throw new Error("Kafka bootstrap server protocol is invalid");
+    }
+    securityProtocol = protocol as KafkaSecurityProtocol;
+    address = schemeMatch[2];
+  } else if (server.includes("://")) {
+    throw new Error("Kafka bootstrap server protocol is invalid");
   }
+
   let parsed: URL;
   try {
-    parsed = new URL(`kafka://${server}`);
+    parsed = new URL(`kafka://${address}`);
   } catch {
     throw new Error("Kafka bootstrap servers are invalid");
   }
   if (!parsed.hostname || parsed.username || parsed.password || parsed.search || parsed.hash || (parsed.pathname && parsed.pathname !== "/")) {
     throw new Error("Kafka bootstrap servers are invalid");
   }
-  return server;
+  if (!parsed.port) {
+    throw new Error("Kafka bootstrap servers must be host:port values");
+  }
+  return { address, securityProtocol };
 }
 
-export function normalizeKafkaBootstrapServers(value: string): string {
-  const servers = requireKafkaBootstrapServers(value)
+export function parseKafkaBootstrapServers(value: string): ParsedKafkaBootstrapServers {
+  const parsedServers = requireKafkaBootstrapServers(value)
     .split(KAFKA_BOOTSTRAP_SERVER_SEPARATOR)
     .map((server) => server.trim())
     .filter(Boolean)
     .map(normalizeKafkaBootstrapServer);
-  if (!servers.length) throw new Error("Kafka bootstrap servers are required");
-  return servers.join(",");
+  if (!parsedServers.length) throw new Error("Kafka bootstrap servers are required");
+
+  const protocols = new Set(parsedServers.map((server) => server.securityProtocol).filter((protocol): protocol is KafkaSecurityProtocol => !!protocol));
+  if (protocols.size > 1) {
+    throw new Error("Kafka bootstrap servers must use one security protocol");
+  }
+
+  const inferredSecurityProtocol = protocols.values().next().value;
+  return {
+    bootstrapServers: parsedServers.map((server) => server.address).join(","),
+    ...(inferredSecurityProtocol ? { inferredSecurityProtocol } : {}),
+  };
+}
+
+export function normalizeKafkaBootstrapServers(value: string): string {
+  return parseKafkaBootstrapServers(value).bootstrapServers;
+}
+
+export function resolveKafkaSecurityProtocol(configured: string, inferred?: KafkaSecurityProtocol): string {
+  return configured.trim() || inferred || "";
 }
diff --git a/apps/desktop/src/lib/connection/mqKafkaConnection.ts b/apps/desktop/src/lib/connection/mqKafkaConnection.ts
new file mode 100644
index 000000000..a735edf67
--- /dev/null
+++ b/apps/desktop/src/lib/connection/mqKafkaConnection.ts
@@ -0,0 +1,83 @@
+import { parseKafkaBootstrapServers, resolveKafkaSecurityProtocol } from "@/lib/connection/kafkaBootstrapServers";
+import { firstZooKeeperEndpoint, normalizeZooKeeperConnectString } from "@/lib/zookeeper/zookeeperConnection";
+
+export type MqKafkaConnectionSource = "bootstrap" | "zookeeper";
+
+export interface MqKafkaConnectionInput {
+  connectionSource: MqKafkaConnectionSource;
+  bootstrapServers: string;
+  zookeeperServers: string;
+  securityProtocol?: string;
+}
+
+export interface MqKafkaConnectionTarget {
+  host: string;
+  port: number;
+  ssl: boolean;
+}
+
+export function resolveMqKafkaConnectionSource(extra: Record<string, unknown>): MqKafkaConnectionSource {
+  if (extra.connectionSource === "zookeeper") return "zookeeper";
+  if (typeof extra.zookeeperServers === "string" && extra.zookeeperServers.trim() && !(typeof extra.bootstrapServers === "string" && extra.bootstrapServers.trim())) {
+    return "zookeeper";
+  }
+  return "bootstrap";
+}
+
+export function normalizeMqKafkaZooKeeperServers(value: string): string {
+  const normalized = normalizeZooKeeperConnectString(value.trim());
+  if (!normalized) throw new Error("Kafka ZooKeeper servers are required");
+
+  const chrootIndex = normalized.indexOf("/");
+  const ensemble = chrootIndex >= 0 ? normalized.slice(0, chrootIndex) : normalized;
+  const chroot = chrootIndex >= 0 ? normalized.slice(chrootIndex) : "";
+  if (chroot && (!chroot.startsWith("/") || chroot.includes("//"))) {
+    throw new Error("Kafka ZooKeeper servers are invalid");
+  }
+
+  for (const endpoint of ensemble.split(",")) {
+    const match = endpoint.match(/^(?:\[[^\]]+\]|[^:\s/?#]+):(\d+)$/u);
+    const port = match ? Number(match[1]) : Number.NaN;
+    if (!match || !Number.isInteger(port) || port < 1 || port > 65535) {
+      throw new Error("Kafka ZooKeeper servers are invalid");
+    }
+  }
+  return normalized;
+}
+
+export function buildMqKafkaConnectionExtra(input: MqKafkaConnectionInput): Record<string, string> {
+  let extra: Record<string, string>;
+  let securityProtocol = input.securityProtocol?.trim() || "";
+
+  if (input.connectionSource === "zookeeper") {
+    extra = {
+      connectionSource: "zookeeper",
+      zookeeperServers: normalizeMqKafkaZooKeeperServers(input.zookeeperServers),
+    };
+  } else {
+    const parsed = parseKafkaBootstrapServers(input.bootstrapServers);
+    securityProtocol = resolveKafkaSecurityProtocol(securityProtocol, parsed.inferredSecurityProtocol);
+    extra = { bootstrapServers: parsed.bootstrapServers };
+  }
+
+  if (securityProtocol) extra.securityProtocol = securityProtocol;
+  return extra;
+}
+
+export function mqKafkaConnectionTarget(input: MqKafkaConnectionInput): MqKafkaConnectionTarget {
+  if (input.connectionSource === "zookeeper") {
+    const endpoint = firstZooKeeperEndpoint(normalizeMqKafkaZooKeeperServers(input.zookeeperServers));
+    if (!endpoint) throw new Error("Kafka ZooKeeper servers are required");
+    return { ...endpoint, ssl: false };
+  }
+
+  const parsed = parseKafkaBootstrapServers(input.bootstrapServers);
+  const first = parsed.bootstrapServers.split(",")[0];
+  const endpoint = new URL(`kafka://${first}`);
+  const securityProtocol = resolveKafkaSecurityProtocol(input.securityProtocol || "", parsed.inferredSecurityProtocol);
+  return {
+    host: endpoint.hostname,
+    port: Number(endpoint.port),
+    ssl: securityProtocol === "SSL" || securityProtocol === "SASL_SSL",
+  };
+}
diff --git a/crates/dbx-core/src/agent_manager.rs b/crates/dbx-core/src/agent_manager.rs
index 610fef396..b0af4965f 100644
--- a/crates/dbx-core/src/agent_manager.rs
+++ b/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() {
diff --git a/crates/dbx-core/src/mq/adapters/kafka.rs b/crates/dbx-core/src/mq/adapters/kafka.rs
index 5d2af1acc..c923f2e11 100644
--- a/crates/dbx-core/src/mq/adapters/kafka.rs
+++ b/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 {
diff --git a/crates/dbx-web/src/routes/mq.rs b/crates/dbx-web/src/routes/mq.rs
index 98685d1c4..d6fc9e6f1 100644
--- a/crates/dbx-web/src/routes/mq.rs
+++ b/crates/dbx-web/src/routes/mq.rs
@@ -1541,6 +1541,7 @@ mod tests {
             sessions: RwLock::new(HashSet::new()),
             sse_channels: RwLock::new(HashMap::new()),
             sql_file_executions: RwLock::new(HashMap::new()),
+            table_import_channels: RwLock::new(HashMap::new()),
             login_rate_limit: Mutex::new(LoginRateLimit { fail_count: 0, locked_until: None }),
             export_files: RwLock::new(HashMap::new()),
         });