From 1317301bcdc5fc519b0e44ccf5ef2166019e1ea3 Mon Sep 17 00:00:00 2001
From: zipg <geeky@foxmail.com>
Date: Thu, 9 Jul 2026 11:41:22 +0800
Subject: [PATCH] feat(kafka): improve topic management and Kerberos auth

---
 .../java/com/dbx/agent/kafka/KafkaAgent.java  |  78 ++++++-
 .../com/dbx/agent/kafka/KafkaAgentTest.java   | 108 +++++++++
 .../connection/ConnectionDialog.vue           |  96 +++++++-
 .../src/components/layout/AppTabBar.vue       |  24 +-
 .../src/components/mq/MonitoringPanel.vue     | 191 +++++++++++++++-
 .../src/components/mq/MqAdminConsole.vue      |   2 +-
 .../src/components/mq/SendMessagePanel.vue    |  19 +-
 .../desktop/src/components/mq/TopicsPanel.vue | 206 +++++++++++++-----
 apps/desktop/src/i18n/locales/en.ts           |  53 +++++
 apps/desktop/src/i18n/locales/es.ts           |  53 +++++
 apps/desktop/src/i18n/locales/it.ts           |  53 +++++
 apps/desktop/src/i18n/locales/ja.ts           |  53 +++++
 apps/desktop/src/i18n/locales/pt-BR.ts        |  53 +++++
 apps/desktop/src/i18n/locales/zh-CN.ts        |  53 +++++
 apps/desktop/src/i18n/locales/zh-TW.ts        |  53 +++++
 apps/desktop/src/lib/connection/mqAuth.ts     |  22 ++
 crates/dbx-core/src/mq/adapters/kafka.rs      |  28 +++
 packages/app-tests/mqAuth.test.ts             |  32 +++
 18 files changed, 1096 insertions(+), 81 deletions(-)
 create mode 100644 apps/desktop/src/lib/connection/mqAuth.ts
 create mode 100644 packages/app-tests/mqAuth.test.ts

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 b53bd3195..ed732e0ca 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
@@ -32,6 +32,13 @@ public final class KafkaAgent {
     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 Set<String> KERBEROS_SYSTEM_PROPERTY_KEYS = Set.of(
+        "java.security.krb5.conf",
+        "sun.security.krb5.debug",
+        "javax.security.auth.useSubjectCredsOnly"
+    );
+    private static final Map<String, String> BASELINE_KERBEROS_SYSTEM_PROPERTIES =
+        snapshotKerberosSystemProperties();
 
     private static final List<String> CAPABILITIES = Collections.unmodifiableList(Arrays.asList(
         "mq_connect", "mq_test_connection", "mq_topics", "mq_consumer_groups",
@@ -138,30 +145,38 @@ public final class KafkaAgent {
 
     private static Object connect(JsonObject params) throws Exception {
         JsonObject conn = connectionObject(params);
-        AdminClient nextAdmin = buildAdminClient(conn);
+        Map<String, String> previousKerberosSystemProperties = applyKerberosSystemProperties(conn);
+        AdminClient nextAdmin = null;
         KafkaProducer<String, byte[]> nextProducer = null;
         try {
+            nextAdmin = buildAdminClient(conn);
             // Verify connectivity
             nextAdmin.describeCluster().clusterId().get(
                 intOrDefault(conn, "request_timeout_ms", DEFAULT_REQUEST_TIMEOUT_MS), TimeUnit.MILLISECONDS);
             nextProducer = buildProducer(conn);
             closeClients();
+            applyKerberosSystemProperties(conn);
             adminClient = nextAdmin;
             producer = nextProducer;
             return Collections.singletonMap("ok", true);
         } catch (Exception e) {
-            nextAdmin.close(Duration.ofSeconds(5));
+            if (nextAdmin != null) {
+                nextAdmin.close(Duration.ofSeconds(5));
+            }
             if (nextProducer != null) {
                 nextProducer.close(Duration.ofSeconds(5));
             }
+            restoreKerberosSystemProperties(previousKerberosSystemProperties);
             throw e;
         }
     }
 
     private static Object testConnection(JsonObject params) throws Exception {
         JsonObject conn = connectionObject(params);
-        AdminClient probe = buildAdminClient(conn);
+        Map<String, String> previousKerberosSystemProperties = applyKerberosSystemProperties(conn);
+        AdminClient probe = null;
         try {
+            probe = buildAdminClient(conn);
             int timeout = intOrDefault(conn, "request_timeout_ms", DEFAULT_REQUEST_TIMEOUT_MS);
             DescribeClusterResult cluster = probe.describeCluster();
             String clusterId = cluster.clusterId().get(timeout, TimeUnit.MILLISECONDS);
@@ -197,7 +212,10 @@ public final class KafkaAgent {
             result.put("brokers", brokerList);
             return result;
         } finally {
-            probe.close(Duration.ofSeconds(5));
+            if (probe != null) {
+                probe.close(Duration.ofSeconds(5));
+            }
+            restoreKerberosSystemProperties(previousKerberosSystemProperties);
         }
     }
 
@@ -210,6 +228,7 @@ public final class KafkaAgent {
             producer.close(Duration.ofSeconds(5));
             producer = null;
         }
+        restoreKerberosSystemProperties(BASELINE_KERBEROS_SYSTEM_PROPERTIES);
     }
 
     // -----------------------------------------------------------------------
@@ -333,12 +352,61 @@ public final class KafkaAgent {
         if (properties != null) {
             for (Map.Entry<String, JsonElement> entry : properties.entrySet()) {
                 if (entry.getValue().isJsonPrimitive()) {
-                    props.put(entry.getKey(), entry.getValue().getAsString());
+                    String key = entry.getKey();
+                    String value = entry.getValue().getAsString();
+                    props.put(key, value);
                 }
             }
         }
     }
 
+    static Map<String, String> applyKerberosSystemProperties(JsonObject conn) {
+        Map<String, String> previous = snapshotKerberosSystemProperties();
+        JsonObject properties = connectionProperties(conn);
+        for (String key : KERBEROS_SYSTEM_PROPERTY_KEYS) {
+            String value = stringProperty(properties, key);
+            if (value == null || value.isBlank()) {
+                value = BASELINE_KERBEROS_SYSTEM_PROPERTIES.get(key);
+            }
+            setOrClearSystemProperty(key, value);
+        }
+        return previous;
+    }
+
+    static void restoreKerberosSystemProperties(Map<String, String> values) {
+        for (String key : KERBEROS_SYSTEM_PROPERTY_KEYS) {
+            setOrClearSystemProperty(key, values.get(key));
+        }
+    }
+
+    private static Map<String, String> snapshotKerberosSystemProperties() {
+        Map<String, String> values = new LinkedHashMap<>();
+        for (String key : KERBEROS_SYSTEM_PROPERTY_KEYS) {
+            values.put(key, System.getProperty(key));
+        }
+        return values;
+    }
+
+    private static JsonObject connectionProperties(JsonObject conn) {
+        return conn.has("properties") && conn.get("properties").isJsonObject()
+            ? conn.getAsJsonObject("properties") : null;
+    }
+
+    private static String stringProperty(JsonObject properties, String key) {
+        if (properties == null || !properties.has(key) || !properties.get(key).isJsonPrimitive()) {
+            return null;
+        }
+        return properties.get(key).getAsString();
+    }
+
+    private static void setOrClearSystemProperty(String key, String value) {
+        if (value == null || value.isBlank()) {
+            System.clearProperty(key);
+        } else {
+            System.setProperty(key, value);
+        }
+    }
+
     // -----------------------------------------------------------------------
     // Topic management
     // -----------------------------------------------------------------------
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 876a78969..1b4ad46db 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
@@ -3,6 +3,9 @@ package com.dbx.agent.kafka;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 
+import com.google.gson.JsonParser;
+import java.util.Map;
+import java.util.Properties;
 import org.junit.jupiter.api.Test;
 
 class KafkaAgentTest {
@@ -30,4 +33,109 @@ class KafkaAgentTest {
     void returnsNoSeekOffsetWhenTopicHasNoReadableMessages() {
         assertNull(KafkaAgent.normalizePeekOffset(0, 5, 5));
     }
+
+    @Test
+    void appliesKerberosKafkaProperties() {
+        Properties props = new Properties();
+        KafkaAgent.applyConnectionProperties(JsonParser.parseString("""
+            {
+              "security_protocol": "SASL_SSL",
+              "sasl_mechanism": "GSSAPI",
+              "properties": {
+                "sasl.jaas.config": "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab=\\"/tmp/user.keytab\\" principal=\\"user@EXAMPLE.COM\\";",
+                "sasl.kerberos.service.name": "kafka"
+              }
+            }
+            """).getAsJsonObject(), props);
+
+        assertEquals("SASL_SSL", props.getProperty("security.protocol"));
+        assertEquals("GSSAPI", props.getProperty("sasl.mechanism"));
+        assertEquals("kafka", props.getProperty("sasl.kerberos.service.name"));
+        assertEquals(
+            "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab=\"/tmp/user.keytab\" principal=\"user@EXAMPLE.COM\";",
+            props.getProperty("sasl.jaas.config")
+        );
+    }
+
+    @Test
+    void appliesAllowedKerberosSystemPropertiesFromConnectionProperties() {
+        Map<String, String> previous = KafkaAgent.applyKerberosSystemProperties(JsonParser.parseString("""
+            {
+              "properties": {
+                "java.security.krb5.conf": "/tmp/krb5.conf",
+                "sun.security.krb5.debug": "true",
+                "custom.system.property": "should-not-leak"
+              }
+            }
+            """).getAsJsonObject());
+        try {
+            assertEquals("/tmp/krb5.conf", System.getProperty("java.security.krb5.conf"));
+            assertEquals("true", System.getProperty("sun.security.krb5.debug"));
+            assertNull(System.getProperty("custom.system.property"));
+        } finally {
+            KafkaAgent.restoreKerberosSystemProperties(previous);
+        }
+    }
+
+    @Test
+    void clearsPreviousKerberosSystemPropertiesForNextConnection() {
+        String baseline = System.getProperty("java.security.krb5.conf");
+        Map<String, String> previous = KafkaAgent.applyKerberosSystemProperties(JsonParser.parseString("""
+            {
+              "properties": {
+                "java.security.krb5.conf": "/tmp/cluster-a.krb5.conf"
+              }
+            }
+            """).getAsJsonObject());
+        try {
+            assertEquals("/tmp/cluster-a.krb5.conf", System.getProperty("java.security.krb5.conf"));
+
+            Map<String, String> beforeSecondConnection = KafkaAgent.applyKerberosSystemProperties(JsonParser.parseString("""
+                {
+                  "properties": {
+                    "sasl.kerberos.service.name": "kafka"
+                  }
+                }
+                """).getAsJsonObject());
+            try {
+                assertEquals(baseline, System.getProperty("java.security.krb5.conf"));
+            } finally {
+                KafkaAgent.restoreKerberosSystemProperties(beforeSecondConnection);
+            }
+        } finally {
+            KafkaAgent.restoreKerberosSystemProperties(previous);
+        }
+    }
+
+    @Test
+    void restoresKerberosSystemPropertiesWhenTestConnectionClientConstructionFails() {
+        String previous = System.getProperty("java.security.krb5.conf");
+        try {
+            String response = KafkaAgent.handleRequest("""
+                {
+                  "jsonrpc": "2.0",
+                  "id": 42,
+                  "method": "test_connection",
+                  "params": {
+                    "connection": {
+                      "bootstrap_servers": "",
+                      "properties": {
+                        "java.security.krb5.conf": "/tmp/leaked-test-connection.krb5.conf"
+                      }
+                    }
+                  }
+                }
+                """);
+
+            assertEquals(-1, JsonParser.parseString(response).getAsJsonObject()
+                .getAsJsonObject("error").get("code").getAsInt());
+            assertEquals(previous, System.getProperty("java.security.krb5.conf"));
+        } finally {
+            if (previous == null) {
+                System.clearProperty("java.security.krb5.conf");
+            } else {
+                System.setProperty("java.security.krb5.conf", previous);
+            }
+        }
+    }
 }
diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue
index 2eff0f04a..21516bffa 100644
--- a/apps/desktop/src/components/connection/ConnectionDialog.vue
+++ b/apps/desktop/src/components/connection/ConnectionDialog.vue
@@ -40,6 +40,7 @@ import { SQLITE_DATABASE_FILE_EXTENSIONS } from "@/lib/database/databaseFileDete
 import { connectionAttemptOriginalErrorMessage, connectionAttemptTimeoutMessage, connectionAttemptTimeoutMs } from "@/lib/connection/connectionAttemptTimeout";
 import { appendConnectionErrorHints } from "@/lib/connection/connectionErrorHints";
 import { normalizeKafkaBootstrapServers } from "@/lib/connection/kafkaBootstrapServers";
+import { detectMqUiAuthKind, isMqAuthKindAllowedForSystem, type MqUiAuthKind } from "@/lib/connection/mqAuth";
 import { driverInstallProgressPercent, type DriverInstallProgress } from "@/lib/connection/driverInstallProgressUi";
 import { ArrowLeft, ArrowDown, ArrowUp, CheckSquare, ChevronRight, CircleHelp, Copy, ExternalLink, FilePlus2, FolderOpen, GripVertical, Grid3X3, KeyRound, Link2, List, ListFilter, Loader2, Pencil, Pipette, Plus, Search, ShieldCheck, Square, Trash2 } from "@lucide/vue";
 import { buildDraftVisibleDatabasesConnectionId, connectionCanChooseVisibleDatabases, initialVisibleDatabaseSelection, visibleDatabaseSelectionIsStale } from "@/lib/connection/connectionVisibleDatabases";
@@ -390,14 +391,17 @@ const dialogStep = ref<DialogStep>("select");
 const dbPickerView = ref<DbPickerView>("icon");
 const dbSearchQuery = ref("");
 const configTab = ref<ConfigTab>("connection");
-type MqAuthKind = MqAuth["kind"];
 const MQ_KAFKA_SECURITY_PROTOCOL_AUTO = "__auto";
 const mqAdminUrl = ref("http://127.0.0.1:8080");
 const mqSystemKind = ref<MqSystemKind>("pulsar");
 const mqKafkaBootstrapServers = ref("127.0.0.1:9092");
 const mqKafkaSecurityProtocol = ref(MQ_KAFKA_SECURITY_PROTOCOL_AUTO);
 const mqKafkaSaslMechanism = ref("PLAIN");
-const mqAuthKind = ref<MqAuthKind>("none");
+const mqKafkaKerberosPrincipal = ref("");
+const mqKafkaKerberosKeytabPath = ref("");
+const mqKafkaKerberosServiceName = ref("kafka");
+const mqKafkaKrb5ConfPath = ref("");
+const mqAuthKind = ref<MqUiAuthKind>("none");
 const mqToken = ref("");
 const mqBasicUsername = ref("");
 const mqBasicPassword = ref("");
@@ -693,18 +697,50 @@ function mqExtraString(extra: Record<string, unknown>, key: string): string {
   return typeof value === "string" ? value : "";
 }
 
+function mqExtraProperties(extra: Record<string, unknown>): Record<string, unknown> {
+  const value = extra.properties;
+  return value && typeof value === "object" && !Array.isArray(value) ? (value as Record<string, unknown>) : {};
+}
+
+function mqExtraPropertyString(extra: Record<string, unknown>, key: string): string {
+  const value = mqExtraProperties(extra)[key];
+  return typeof value === "string" ? value : "";
+}
+
+function jaasStringValue(value: string): string {
+  return value.replaceAll("\\", "\\\\").replaceAll('"', '\\"');
+}
+
+function parseJaasStringProperty(value: string, key: string): string {
+  const escapedKey = key.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+  const match = value.match(new RegExp(`${escapedKey}\\s*=\\s*"((?:\\\\.|[^"\\\\])*)"`, "i"));
+  if (!match) return "";
+  return match[1].replace(/\\(["\\])/g, "$1");
+}
+
 function resetMqFields(config?: Partial<MqAdminConfig>) {
   const systemKind = config?.systemKind === "kafka" ? "kafka" : "pulsar";
   const extra = mqExtraRecord(config);
+  const properties = mqExtraProperties(extra);
+  const jaasConfig = mqExtraPropertyString(extra, "sasl.jaas.config");
   mqSystemKind.value = systemKind;
   mqAdminUrl.value = config?.adminUrl?.trim() || (systemKind === "kafka" ? "" : "http://127.0.0.1:8080");
   mqKafkaBootstrapServers.value = mqExtraString(extra, "bootstrapServers") || "127.0.0.1:9092";
   mqKafkaSecurityProtocol.value = mqExtraString(extra, "securityProtocol") || MQ_KAFKA_SECURITY_PROTOCOL_AUTO;
   mqKafkaSaslMechanism.value = mqExtraString(extra, "saslMechanism") || "PLAIN";
+  mqKafkaKerberosPrincipal.value = parseJaasStringProperty(jaasConfig, "principal");
+  mqKafkaKerberosKeytabPath.value = parseJaasStringProperty(jaasConfig, "keyTab");
+  mqKafkaKerberosServiceName.value = typeof properties["sasl.kerberos.service.name"] === "string" ? properties["sasl.kerberos.service.name"] : "kafka";
+  mqKafkaKrb5ConfPath.value = typeof properties["java.security.krb5.conf"] === "string" ? properties["java.security.krb5.conf"] : "";
   mqTlsSkipVerify.value = !!config?.tlsSkipVerify;
   mqPinnedVersion.value = pinnedVersionToSelection(config?.pinnedVersion);
   const auth = (config?.auth || { kind: "none" }) as MqAuth;
-  mqAuthKind.value = systemKind === "kafka" && auth.kind !== "basic" ? "none" : auth.kind || "none";
+  mqAuthKind.value = detectMqUiAuthKind({
+    systemKind,
+    authKind: auth.kind,
+    saslMechanism: mqKafkaSaslMechanism.value,
+    jaasConfig,
+  });
   mqToken.value = auth.token || "";
   mqBasicUsername.value = auth.username || "";
   mqBasicPassword.value = auth.password || "";
@@ -741,12 +777,18 @@ function hydrateMqFields(value: unknown) {
 watch(mqSystemKind, (kind) => {
   if (kind === "kafka") {
     if (!mqKafkaBootstrapServers.value.trim()) mqKafkaBootstrapServers.value = "127.0.0.1:9092";
-    if (!["none", "basic"].includes(mqAuthKind.value)) mqAuthKind.value = "none";
+    if (!isMqAuthKindAllowedForSystem(kind, mqAuthKind.value)) mqAuthKind.value = "none";
     return;
   }
   if (!mqAdminUrl.value.trim()) mqAdminUrl.value = "http://127.0.0.1:8080";
 });
 
+watch(mqAuthKind, (kind) => {
+  if (mqSystemKind.value === "kafka" && kind === "basic" && mqKafkaSaslMechanism.value.toUpperCase() === "GSSAPI") {
+    mqKafkaSaslMechanism.value = "PLAIN";
+  }
+});
+
 function resetNacosFields(config?: Partial<NacosAdminConfig>) {
   nacosServerAddr.value = config?.serverAddr?.trim() || NACOS_DEFAULT_CONSOLE_URL;
   nacosNamespace.value = config?.namespace || "";
@@ -845,6 +887,12 @@ function buildMqAuth(): MqAuth {
   }
 }
 
+function buildKafkaKerberosJaasConfig(): string {
+  const principal = requireMqField(mqKafkaKerberosPrincipal.value, t("connection.kafkaKerberosPrincipalRequired"));
+  const keytab = requireMqField(mqKafkaKerberosKeytabPath.value, t("connection.kafkaKerberosKeytabRequired"));
+  return `com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="${jaasStringValue(keytab)}" principal="${jaasStringValue(principal)}";`;
+}
+
 function buildMqTokenSigning() {
   if (mqTokenSigningMode.value === "none") return undefined;
   return {
@@ -857,11 +905,21 @@ function buildMqAdminConfig(): MqAdminConfig {
   const systemKind = mqSystemKind.value;
   if (systemKind === "kafka") {
     const bootstrapServers = normalizeKafkaBootstrapServers(mqKafkaBootstrapServers.value);
-    const extra: Record<string, string> = { bootstrapServers };
+    const extra: Record<string, unknown> = { bootstrapServers };
     const securityProtocol = mqKafkaSecurityProtocol.value === MQ_KAFKA_SECURITY_PROTOCOL_AUTO ? "" : mqKafkaSecurityProtocol.value.trim();
-    const saslMechanism = mqKafkaSaslMechanism.value.trim();
+    const saslMechanism = mqAuthKind.value === "kerberos" ? "GSSAPI" : mqKafkaSaslMechanism.value.trim();
+    const properties: Record<string, string> = {};
     if (securityProtocol) extra.securityProtocol = securityProtocol;
     if (mqAuthKind.value === "basic" && saslMechanism) extra.saslMechanism = saslMechanism;
+    if (mqAuthKind.value === "kerberos") {
+      extra.saslMechanism = "GSSAPI";
+      properties["sasl.jaas.config"] = buildKafkaKerberosJaasConfig();
+      properties["sasl.kerberos.service.name"] = mqKafkaKerberosServiceName.value.trim() || "kafka";
+      if (mqKafkaKrb5ConfPath.value.trim()) {
+        properties["java.security.krb5.conf"] = mqKafkaKrb5ConfPath.value.trim();
+      }
+    }
+    if (Object.keys(properties).length) extra.properties = properties;
     return {
       systemKind: mqSystemKind.value,
       adminUrl: "",
@@ -3908,6 +3966,7 @@ function openExternalUrl(url: string) {
                       <Button size="sm" :variant="mqAuthKind === 'none' ? 'default' : 'outline'" @click="mqAuthKind = 'none'">None</Button>
                       <Button v-if="mqSystemKind !== 'kafka'" size="sm" :variant="mqAuthKind === 'token' ? 'default' : 'outline'" @click="mqAuthKind = 'token'">Token</Button>
                       <Button size="sm" :variant="mqAuthKind === 'basic' ? 'default' : 'outline'" @click="mqAuthKind = 'basic'">Basic</Button>
+                      <Button v-if="mqSystemKind === 'kafka'" size="sm" :variant="mqAuthKind === 'kerberos' ? 'default' : 'outline'" @click="mqAuthKind = 'kerberos'">Kerberos</Button>
                       <Button v-if="mqSystemKind !== 'kafka'" size="sm" :variant="mqAuthKind === 'apiKey' ? 'default' : 'outline'" @click="mqAuthKind = 'apiKey'">API Key</Button>
                       <Button v-if="mqSystemKind !== 'kafka'" size="sm" :variant="mqAuthKind === 'oauth2' ? 'default' : 'outline'" @click="mqAuthKind = 'oauth2'">OAuth2</Button>
                     </div>
@@ -3941,6 +4000,31 @@ function openExternalUrl(url: string) {
                       </Select>
                     </div>
                   </template>
+                  <template v-else-if="mqSystemKind === 'kafka' && mqAuthKind === 'kerberos'">
+                    <div class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.kafkaKerberosPrincipal") }}</Label>
+                      <Input v-model="mqKafkaKerberosPrincipal" class="col-span-3" placeholder="user@EXAMPLE.COM" />
+                    </div>
+                    <div class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.kafkaKerberosKeytab") }}</Label>
+                      <Input v-model="mqKafkaKerberosKeytabPath" class="col-span-3" :placeholder="t('connection.kafkaKerberosKeytabPlaceholder')" />
+                    </div>
+                    <div class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.kafkaKerberosServiceName") }}</Label>
+                      <Input v-model="mqKafkaKerberosServiceName" class="col-span-3" placeholder="kafka" />
+                    </div>
+                    <div class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.kafkaKerberosKrb5Conf") }}</Label>
+                      <Input v-model="mqKafkaKrb5ConfPath" class="col-span-3" :placeholder="t('connection.kafkaKerberosKrb5ConfPlaceholder')" />
+                    </div>
+                    <div class="grid grid-cols-4 items-start gap-4">
+                      <div></div>
+                      <div class="col-span-3 space-y-1 text-xs leading-5 text-muted-foreground">
+                        <p>{{ t("connection.kafkaKerberosPathHint") }}</p>
+                        <p>{{ t("connection.kafkaKerberosAuthHint") }}</p>
+                      </div>
+                    </div>
+                  </template>
                   <template v-else-if="mqAuthKind === 'apiKey'">
                     <div class="grid grid-cols-4 items-center gap-4">
                       <Label :class="connectionLabelClass">Header</Label>
diff --git a/apps/desktop/src/components/layout/AppTabBar.vue b/apps/desktop/src/components/layout/AppTabBar.vue
index c9fb781cf..3f3002890 100644
--- a/apps/desktop/src/components/layout/AppTabBar.vue
+++ b/apps/desktop/src/components/layout/AppTabBar.vue
@@ -8,6 +8,8 @@ import { Tooltip, TooltipTrigger, TooltipContent } from "@/components/ui/tooltip
 import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover";
 import { Dialog, DialogContent, DialogFooter, DialogHeader, DialogTitle } from "@/components/ui/dialog";
 import { Button } from "@/components/ui/button";
+import DatabaseIcon from "@/components/icons/DatabaseIcon.vue";
+import { useConnectionStore } from "@/stores/connectionStore";
 import { useQueryStore } from "@/stores/queryStore";
 import { useSettingsStore } from "@/stores/settingsStore";
 import { useTabScroll } from "@/composables/useTabScroll";
@@ -40,6 +42,7 @@ const emit = defineEmits<{
 }>();
 
 const { t } = useI18n();
+const connectionStore = useConnectionStore();
 const queryStore = useQueryStore();
 const settingsStore = useSettingsStore();
 const { toast } = useToast();
@@ -414,10 +417,23 @@ function tabColorStyle(tab: QueryTab) {
 }
 
 function tabIconClass(tab: QueryTab) {
+  if (tab.mode === "mq") return "";
   if (tab.mode === "data" || tab.mode === "mongo" || tab.mode === "vector" || tab.mode === "redis" || tab.mode === "objects" || tab.mode === "structure") return "text-emerald-600 dark:text-emerald-400";
   return "text-blue-600 dark:text-blue-400";
 }
 
+function tabDatabaseIconType(tab: QueryTab) {
+  const connection = connectionStore.getConfig(tab.connectionId);
+  if (!connection) return "mq";
+  if (connection.db_type === "mq") {
+    const externalConfig = connection.external_config as { systemKind?: unknown } | undefined;
+    const systemKind = typeof externalConfig?.systemKind === "string" ? externalConfig.systemKind : "";
+    if (connection.driver_profile === "kafka" || systemKind === "kafka") return "kafka";
+    if (connection.driver_profile === "pulsar" || systemKind === "pulsar") return "pulsar";
+  }
+  return connection.driver_profile || connection.db_type;
+}
+
 const showRegularTabScrollbar = computed(() => hasTabOverflow.value);
 const showFixedTabScrollbar = computed(() => hasFixedTabOverflow.value);
 const showRegularTabOverflowControls = computed(() => regularTabs.value.length > 0 && hasTabOverflow.value);
@@ -561,6 +577,7 @@ function onOverflowItemKeydown(event: KeyboardEvent, tabId: string, kind: "regul
                   >
                     <span class="shrink-0" :class="tabIconClass(tab)">
                       <Table2 v-if="tab.mode === 'data' || tab.mode === 'mongo' || tab.mode === 'redis'" class="h-3.5 w-3.5" />
+                      <DatabaseIcon v-else-if="tab.mode === 'mq'" :db-type="tabDatabaseIconType(tab)" class="h-3.5 w-3.5" />
                       <TableProperties v-else-if="tab.mode === 'vector'" class="h-3.5 w-3.5" />
                       <KeyRound v-else-if="tab.mode === 'etcd' || tab.mode === 'zookeeper'" class="h-3.5 w-3.5" />
                       <Network v-else-if="tab.mode === 'nacos'" class="h-3.5 w-3.5" />
@@ -682,7 +699,8 @@ function onOverflowItemKeydown(event: KeyboardEvent, tabId: string, kind: "regul
                 @contextmenu="onContextMenu"
                 @keydown="onOverflowItemKeydown($event, tab.id, 'regular')"
               >
-                <component :is="tabMenuIcon(tab)" :class="['h-3.5 w-3.5 shrink-0', tabIconClass(tab)]" />
+                <DatabaseIcon v-if="tab.mode === 'mq'" :db-type="tabDatabaseIconType(tab)" class="h-3.5 w-3.5 shrink-0" />
+                <component :is="tabMenuIcon(tab)" v-else :class="['h-3.5 w-3.5 shrink-0', tabIconClass(tab)]" />
                 <span class="inline-flex min-w-0 flex-1 items-center gap-0.5 overflow-hidden">
                   <span v-if="isDirtyTab(tab)" aria-hidden="true" class="dirty-tab-marker">*</span>
                   <span class="min-w-0 flex-1 truncate" :style="tabTitleStyle(tab)">{{ tabTitleText(tab) }}</span>
@@ -741,6 +759,7 @@ function onOverflowItemKeydown(event: KeyboardEvent, tabId: string, kind: "regul
                   >
                     <span class="shrink-0" :class="tabIconClass(tab)">
                       <Table2 v-if="tab.mode === 'data' || tab.mode === 'mongo' || tab.mode === 'redis'" class="h-3.5 w-3.5" />
+                      <DatabaseIcon v-else-if="tab.mode === 'mq'" :db-type="tabDatabaseIconType(tab)" class="h-3.5 w-3.5" />
                       <TableProperties v-else-if="tab.mode === 'vector'" class="h-3.5 w-3.5" />
                       <KeyRound v-else-if="tab.mode === 'etcd' || tab.mode === 'zookeeper'" class="h-3.5 w-3.5" />
                       <Network v-else-if="tab.mode === 'nacos'" class="h-3.5 w-3.5" />
@@ -810,7 +829,8 @@ function onOverflowItemKeydown(event: KeyboardEvent, tabId: string, kind: "regul
                 @contextmenu="onContextMenu"
                 @keydown="onOverflowItemKeydown($event, tab.id, 'fixed')"
               >
-                <component :is="tabMenuIcon(tab)" :class="['h-3.5 w-3.5 shrink-0', tabIconClass(tab)]" />
+                <DatabaseIcon v-if="tab.mode === 'mq'" :db-type="tabDatabaseIconType(tab)" class="h-3.5 w-3.5 shrink-0" />
+                <component :is="tabMenuIcon(tab)" v-else :class="['h-3.5 w-3.5 shrink-0', tabIconClass(tab)]" />
                 <span class="inline-flex min-w-0 flex-1 items-center gap-0.5 overflow-hidden">
                   <span v-if="isDirtyTab(tab)" aria-hidden="true" class="dirty-tab-marker">*</span>
                   <span class="min-w-0 flex-1 truncate" :style="tabTitleStyle(tab)">{{ tabTitleText(tab) }}</span>
diff --git a/apps/desktop/src/components/mq/MonitoringPanel.vue b/apps/desktop/src/components/mq/MonitoringPanel.vue
index 29f4c9e10..966f6f5af 100644
--- a/apps/desktop/src/components/mq/MonitoringPanel.vue
+++ b/apps/desktop/src/components/mq/MonitoringPanel.vue
@@ -7,8 +7,8 @@ import { LineChart } from "echarts/charts";
 import { GridComponent, LegendComponent, TooltipComponent } from "echarts/components";
 import VChart from "vue-echarts";
 import { Activity, AlertTriangle, BarChart3, Boxes, CheckCircle2, Database, Download, Gauge, Hash, HardDrive, Layers3, Loader2, Package, RadioTower, RefreshCw, Send, ShieldCheck, Table2, Upload, Users } from "@lucide/vue";
-import type { TopicRef, TopicInfo, TopicStats, BacklogStats } from "@/types/mq";
-import { mqGetTopicStats, mqGetBacklog } from "@/lib/backend/api";
+import type { TopicRef, TopicInfo, TopicStats, BacklogStats, PeekedMessage } from "@/types/mq";
+import { mqGetTopicStats, mqGetBacklog, mqPeekMessages } from "@/lib/backend/api";
 
 use([CanvasRenderer, LineChart, GridComponent, LegendComponent, TooltipComponent]);
 
@@ -62,6 +62,10 @@ const error = ref<string>();
 const autoRefresh = ref(true);
 const refreshInterval = ref(5); // seconds
 const selectedPartitionName = ref<string>();
+const kafkaMessageSql = ref("");
+const kafkaMessageLoading = ref(false);
+const kafkaMessageError = ref<string>();
+const kafkaMessages = ref<PeekedMessage[]>([]);
 
 let refreshTimer: number | undefined;
 const history = ref<MetricPoint[]>([]);
@@ -194,6 +198,61 @@ function refreshNow() {
   void loadStats();
 }
 
+function defaultKafkaMessageSql(): string {
+  const topic = props.topic?.shortName;
+  return topic ? `SELECT * FROM "${topic}" PARTITION 0 OFFSET 0 LIMIT 20` : "";
+}
+
+function parseKafkaMessageSql(sql: string): { topic: string; partition: number; offset: number; limit: number } {
+  const match = sql.trim().match(/^\s*select\s+\*\s+from\s+(?:"([^"]+)"|`([^`]+)`|'([^']+)'|([^\s;]+))(?:\s+partition\s+(\d+))?(?:\s+offset\s+(\d+))?(?:\s+limit\s+(\d+))?\s*;?\s*$/i);
+  if (!match) {
+    throw new Error('仅支持 SELECT * FROM "topic" [PARTITION n] [OFFSET n] [LIMIT n]');
+  }
+  const topic = match[1] || match[2] || match[3] || match[4] || "";
+  const partition = Math.max(0, Number(match[5] ?? 0));
+  const offset = Math.max(0, Number(match[6] ?? 0));
+  const limit = Math.max(1, Math.min(100, Number(match[7] ?? 20)));
+  return { topic, partition, offset, limit };
+}
+
+async function runKafkaMessageSql() {
+  if (!props.tenant || !props.namespace) return;
+  kafkaMessageLoading.value = true;
+  kafkaMessageError.value = undefined;
+  try {
+    const parsed = parseKafkaMessageSql(kafkaMessageSql.value);
+    const selected = props.topic && parsed.topic === props.topic.shortName ? props.topic : undefined;
+    kafkaMessages.value = await mqPeekMessages(
+      props.connectionId,
+      {
+        tenant: props.tenant,
+        namespace: props.namespace,
+        topic: parsed.topic,
+        persistent: selected?.persistent ?? true,
+        partitioned: selected?.partitioned,
+      },
+      "__dbx_kafka_monitor__",
+      parsed.limit,
+      { partition: parsed.partition, offset: parsed.offset },
+    );
+  } catch (e: unknown) {
+    kafkaMessageError.value = formatError(e);
+  } finally {
+    kafkaMessageLoading.value = false;
+  }
+}
+
+function kafkaMessagePayload(message: PeekedMessage): string {
+  return message.payloadText ?? message.payloadBase64;
+}
+
+function formatKafkaMessageTimestamp(value?: string): string {
+  if (!value) return "-";
+  const numeric = Number(value);
+  if (!Number.isFinite(numeric)) return value;
+  return new Date(numeric).toLocaleString();
+}
+
 function appendHistoryPoint(statsData: TopicStats, backlogData?: BacklogStats) {
   const point: MetricPoint = {
     time: new Date().toLocaleTimeString(),
@@ -349,6 +408,9 @@ watch(
   () => {
     history.value = [];
     selectedPartitionName.value = undefined;
+    kafkaMessageSql.value = defaultKafkaMessageSql();
+    kafkaMessageError.value = undefined;
+    kafkaMessages.value = [];
     void loadStats();
     startAutoRefresh();
   },
@@ -534,6 +596,38 @@ onUnmounted(() => {
         </div>
         <div v-else class="empty-state compact">当前 Kafka 响应未返回分区指标</div>
       </div>
+
+      <div class="stats-section">
+        <div class="section-title-row">
+          <h4>Kafka 消息查询</h4>
+          <button type="button" class="btn-sm" :disabled="kafkaMessageLoading || !kafkaMessageSql.trim()" @click="runKafkaMessageSql">
+            <Loader2 v-if="kafkaMessageLoading" class="btn-icon spinning" :size="14" />
+            <span>{{ kafkaMessageLoading ? "查询中..." : "查询消息" }}</span>
+          </button>
+        </div>
+        <textarea v-model="kafkaMessageSql" class="kafka-sql-input" rows="2" spellcheck="false" />
+        <div class="query-hint">支持：SELECT * FROM "topic" PARTITION 0 OFFSET 0 LIMIT 20，单次最多返回 100 条。</div>
+        <div v-if="kafkaMessageError" class="panel-error inline-error">
+          <AlertTriangle :size="16" />
+          <span>{{ kafkaMessageError }}</span>
+        </div>
+        <div v-else-if="kafkaMessageLoading" class="empty-state compact">消息加载中...</div>
+        <div v-else-if="!kafkaMessages.length" class="empty-state compact">暂无消息</div>
+        <div v-else class="kafka-message-list">
+          <article v-for="message in kafkaMessages" :key="message.messageId || message.position" class="kafka-message-row">
+            <div class="kafka-message-meta">
+              <span>#{{ message.position }}</span>
+              <span>offset {{ message.messageId || "-" }}</span>
+              <span v-if="message.key">key {{ message.key }}</span>
+              <span>{{ formatKafkaMessageTimestamp(message.publishTime) }}</span>
+            </div>
+            <pre class="kafka-message-payload">{{ kafkaMessagePayload(message) }}</pre>
+            <div v-if="Object.keys(message.headers || {}).length" class="kafka-message-headers">
+              <span v-for="(value, key) in message.headers" :key="key">{{ key }}: {{ value }}</span>
+            </div>
+          </article>
+        </div>
+      </div>
     </div>
 
     <div v-else-if="stats" class="stats-container">
@@ -1059,6 +1153,99 @@ onUnmounted(() => {
   box-shadow: 0 0 0 4px var(--monitor-accent-soft);
 }
 
+.section-title-row {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 12px;
+  margin-bottom: 12px;
+}
+
+.section-title-row h4 {
+  margin-bottom: 0;
+}
+
+.kafka-sql-input {
+  width: 100%;
+  min-height: 54px;
+  padding: 10px 12px;
+  border: 1px solid var(--monitor-border);
+  border-radius: 8px;
+  background: var(--monitor-surface);
+  color: var(--monitor-text);
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
+  font-size: 12px;
+  line-height: 1.5;
+  resize: vertical;
+  box-sizing: border-box;
+}
+
+.kafka-sql-input:focus {
+  outline: none;
+  border-color: var(--monitor-accent);
+  box-shadow: 0 0 0 3px var(--monitor-accent-soft);
+}
+
+.query-hint {
+  margin-top: 6px;
+  color: var(--monitor-faint);
+  font-size: 12px;
+}
+
+.inline-error {
+  justify-content: flex-start;
+  margin-top: 10px;
+  border: 1px solid color-mix(in srgb, var(--monitor-danger) 22%, transparent);
+  border-radius: 8px;
+  padding: 10px 12px;
+}
+
+.kafka-message-list {
+  display: grid;
+  gap: 10px;
+  margin-top: 12px;
+}
+
+.kafka-message-row {
+  border: 1px solid var(--monitor-border);
+  border-radius: 8px;
+  background: var(--monitor-surface);
+  overflow: hidden;
+}
+
+.kafka-message-meta {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 10px;
+  padding: 8px 10px;
+  border-bottom: 1px solid var(--monitor-border);
+  color: var(--monitor-muted);
+  font-size: 12px;
+}
+
+.kafka-message-payload {
+  max-height: 220px;
+  margin: 0;
+  overflow: auto;
+  padding: 10px;
+  color: var(--monitor-text);
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
+  font-size: 12px;
+  line-height: 1.5;
+  white-space: pre-wrap;
+  word-break: break-word;
+}
+
+.kafka-message-headers {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 6px;
+  padding: 8px 10px;
+  border-top: 1px solid var(--monitor-border);
+  color: var(--monitor-muted);
+  font-size: 11px;
+}
+
 .charts-grid {
   display: grid;
   grid-template-columns: repeat(auto-fit, minmax(280px, 1fr));
diff --git a/apps/desktop/src/components/mq/MqAdminConsole.vue b/apps/desktop/src/components/mq/MqAdminConsole.vue
index 587c18b53..6585075fc 100644
--- a/apps/desktop/src/components/mq/MqAdminConsole.vue
+++ b/apps/desktop/src/components/mq/MqAdminConsole.vue
@@ -129,7 +129,7 @@ function handleNamespaceRolesSelected(namespace: string) {
 function handleTopicSelected(topic: TopicInfo) {
   selectedTopic.value = topic;
   selectedSubscriptionName.value = undefined;
-  activeTab.value = canManageSubscriptions.value ? "subscriptions" : "monitoring";
+  activeTab.value = isKafkaCluster.value ? "monitoring" : canManageSubscriptions.value ? "subscriptions" : "monitoring";
 }
 
 function handleSubscriptionSelected(subscription: string) {
diff --git a/apps/desktop/src/components/mq/SendMessagePanel.vue b/apps/desktop/src/components/mq/SendMessagePanel.vue
index d65ab3742..2fae77e4e 100644
--- a/apps/desktop/src/components/mq/SendMessagePanel.vue
+++ b/apps/desktop/src/components/mq/SendMessagePanel.vue
@@ -59,6 +59,7 @@ const selectedTopicRef = computed<TopicRef | null>(() => {
 });
 
 const canBrowseMessages = computed(() => props.isKafkaCluster === true && props.supportsPeekMessages !== false);
+const topicListId = computed(() => `mq-topic-options-${props.connectionId}`);
 
 function clearSuccessLater() {
   if (successTimer) clearTimeout(successTimer);
@@ -253,18 +254,17 @@ watch(
       <div class="form-group">
         <label>目标主题 <span class="required">*</span></label>
         <div class="topic-select-row">
-          <select v-model="topicName" :disabled="readOnly || topicsLoading" class="topic-select">
-            <option value="" disabled>{{ topicsLoading ? "加载中..." : "选择主题..." }}</option>
-            <option v-for="t in topicOptions" :key="t.value" :value="t.value">
-              {{ t.label }}<template v-if="t.partitions != null"> ({{ t.partitions }} 分区)</template>
-            </option>
-          </select>
+          <input v-model="topicName" :list="topicListId" :disabled="readOnly || topicsLoading" class="topic-input" :placeholder="topicsLoading ? '加载中...' : '输入或搜索主题...'" autocomplete="off" />
+          <datalist :id="topicListId">
+            <option v-for="t in topicOptions" :key="t.value" :value="t.value" :label="t.partitions != null ? `${t.label} (${t.partitions} 分区)` : t.label" />
+          </datalist>
           <button @click="loadTopics" :disabled="topicsLoading" class="btn-icon" title="刷新主题列表">
             <span v-if="topicsLoading" class="spin">⟳</span>
             <span v-else>⟳</span>
           </button>
         </div>
         <div v-if="!availableTopics.length && !topicsLoading" class="form-hint">暂无可用主题</div>
+        <div v-else class="form-hint">可输入关键词搜索，也可以直接粘贴 topic 名称。</div>
       </div>
 
       <!-- 消息键 -->
@@ -462,7 +462,7 @@ watch(
   gap: 8px;
 }
 
-.topic-select {
+.topic-input {
   flex: 1;
   padding: 7px 10px;
   border: 1px solid var(--color-border);
@@ -470,11 +470,10 @@ watch(
   background: var(--color-background);
   color: var(--color-text);
   font-size: 13px;
-  cursor: pointer;
-  appearance: auto;
+  min-width: 0;
 }
 
-.topic-select:focus {
+.topic-input:focus {
   outline: none;
   border-color: var(--color-primary);
   box-shadow: 0 0 0 2px var(--color-primary-alpha);
diff --git a/apps/desktop/src/components/mq/TopicsPanel.vue b/apps/desktop/src/components/mq/TopicsPanel.vue
index 286517792..41ae42424 100644
--- a/apps/desktop/src/components/mq/TopicsPanel.vue
+++ b/apps/desktop/src/components/mq/TopicsPanel.vue
@@ -1,5 +1,6 @@
 <script setup lang="ts">
 import { ref, watch, computed } from "vue";
+import { useI18n } from "vue-i18n";
 import type { NamespaceRef, TopicRef, TopicInfo, ListTopicsOpts } from "@/types/mq";
 import { mqListTopics, mqCreateTopic, mqDeleteTopic, mqUpdatePartitions } from "@/lib/backend/api";
 import { formatError } from "@/lib/backend/errorUtils";
@@ -18,13 +19,17 @@ const emit = defineEmits<{
   topicSelected: [topic: TopicInfo];
 }>();
 
+const { t } = useI18n();
+
 const topics = ref<TopicInfo[]>([]);
 const loading = ref(false);
 const error = ref<string>();
+const dialogError = ref<string>();
 const showCreateDialog = ref(false);
 const showPartitionsDialog = ref(false);
 const selectedTopic = ref<TopicInfo>();
 const editingTopic = ref<TopicInfo>();
+const topicSearch = ref("");
 
 const formData = ref({
   topicName: "",
@@ -36,15 +41,23 @@ const formData = ref({
 const newPartitions = ref(4);
 
 const includeNonPersistent = ref(false);
-const readOnlyMessage = "当前连接为只读模式，不能执行写操作";
 
 const filteredTopics = computed(() => {
-  return topics.value;
+  const query = topicSearch.value.trim().toLowerCase();
+  if (!query) return topics.value;
+  return topics.value.filter((topic) => {
+    return topic.name.toLowerCase().includes(query) || topic.shortName.toLowerCase().includes(query);
+  });
+});
+const editingCurrentPartitions = computed(() => editingTopic.value?.partitions ?? 0);
+const canSubmitPartitionUpdate = computed(() => {
+  const current = editingCurrentPartitions.value;
+  return !props.readOnly && current > 0 && Number.isFinite(newPartitions.value) && newPartitions.value > current;
 });
 
 function guardWritable() {
   if (props.readOnly) {
-    error.value = readOnlyMessage;
+    error.value = t("mqTopics.readOnly");
     return false;
   }
   return true;
@@ -75,6 +88,7 @@ async function loadTopics() {
 
 function openCreateDialog() {
   if (!guardWritable()) return;
+  dialogError.value = undefined;
   formData.value = {
     topicName: "",
     persistent: true,
@@ -86,8 +100,9 @@ function openCreateDialog() {
 
 function openPartitionsDialog(topic: TopicInfo) {
   if (!guardWritable()) return;
+  dialogError.value = undefined;
   if (!topic.partitions || topic.partitions < 1) {
-    error.value = "当前分区数未知，无法安全调整分区";
+    error.value = t("mqTopics.currentPartitionsUnknown");
     return;
   }
   editingTopic.value = topic;
@@ -98,7 +113,7 @@ function openPartitionsDialog(topic: TopicInfo) {
 async function handleCreate() {
   if (!guardWritable()) return;
   if (!formData.value.topicName.trim() || !props.tenant || !props.namespace) {
-    error.value = "Topic name is required";
+    dialogError.value = t("mqTopics.topicNameRequired");
     return;
   }
   loading.value = true;
@@ -113,9 +128,10 @@ async function handleCreate() {
     const partitions = props.supportsPartitionedTopics !== false && formData.value.partitioned ? formData.value.partitions : undefined;
     await mqCreateTopic(props.connectionId, topicRef, partitions);
     showCreateDialog.value = false;
+    dialogError.value = undefined;
     await loadTopics();
   } catch (e: unknown) {
-    error.value = formatError(e);
+    dialogError.value = formatError(e);
   } finally {
     loading.value = false;
   }
@@ -123,7 +139,7 @@ async function handleCreate() {
 
 async function handleDelete(topic: TopicInfo) {
   if (!guardWritable()) return;
-  if (!confirm(`确定要删除主题 "${topic.shortName}" 吗？此操作不可逆。`)) return;
+  if (!confirm(t("mqTopics.confirmDelete", { name: topic.shortName }))) return;
   if (!props.tenant || !props.namespace) return;
   loading.value = true;
   error.value = undefined;
@@ -151,11 +167,11 @@ async function handleUpdatePartitions() {
   if (!editingTopic.value || !props.tenant || !props.namespace) return;
   const currentPartitions = editingTopic.value.partitions;
   if (!currentPartitions || currentPartitions < 1) {
-    error.value = "当前分区数未知，无法安全调整分区";
+    dialogError.value = t("mqTopics.currentPartitionsUnknown");
     return;
   }
   if (newPartitions.value <= currentPartitions) {
-    error.value = "新分区数必须大于当前分区数";
+    dialogError.value = t("mqTopics.partitionMustIncrease");
     return;
   }
   loading.value = true;
@@ -169,9 +185,10 @@ async function handleUpdatePartitions() {
     };
     await mqUpdatePartitions(props.connectionId, topicRef, newPartitions.value);
     showPartitionsDialog.value = false;
+    dialogError.value = undefined;
     await loadTopics();
   } catch (e: unknown) {
-    error.value = formatError(e);
+    dialogError.value = formatError(e);
   } finally {
     loading.value = false;
   }
@@ -182,6 +199,14 @@ function selectTopic(topic: TopicInfo) {
   emit("topicSelected", topic);
 }
 
+function normalizePartitionInput() {
+  const min = editingCurrentPartitions.value + 1;
+  if (!showPartitionsDialog.value || min <= 1) return;
+  if (!Number.isFinite(Number(newPartitions.value)) || Number(newPartitions.value) < min) {
+    newPartitions.value = min;
+  }
+}
+
 watch(
   () => [props.tenant, props.namespace],
   () => {
@@ -194,42 +219,52 @@ watch(
 watch(includeNonPersistent, () => {
   loadTopics();
 });
+
+watch(newPartitions, () => {
+  if (dialogError.value === t("mqTopics.partitionMustIncrease") && canSubmitPartitionUpdate.value) {
+    dialogError.value = undefined;
+  }
+});
 </script>
 
 <template>
   <div class="topics-panel">
     <div class="panel-toolbar">
       <div class="toolbar-left">
-        <h3>主题管理</h3>
+        <h3>{{ t("mqTopics.title") }}</h3>
+        <input v-model="topicSearch" type="search" class="topic-search" :placeholder="t('mqTopics.searchPlaceholder')" :disabled="loading && !topics.length" />
+        <span v-if="topics.length" class="topic-count">{{ filteredTopics.length }} / {{ topics.length }}</span>
         <label class="checkbox-label">
           <input type="checkbox" v-model="includeNonPersistent" />
-          包含非持久化主题
+          {{ t("mqTopics.includeNonPersistent") }}
         </label>
       </div>
       <div class="toolbar-actions">
         <button @click="loadTopics" :disabled="loading || !tenant || !namespace" class="btn-secondary">
-          {{ loading ? "刷新中..." : "刷新" }}
+          {{ loading ? t("mqTopics.refreshing") : t("mqTopics.refresh") }}
         </button>
-        <button @click="openCreateDialog" :disabled="loading || readOnly || !tenant || !namespace" class="btn-primary">+ 创建主题</button>
+        <button @click="openCreateDialog" :disabled="loading || readOnly || !tenant || !namespace" class="btn-primary">+ {{ t("mqTopics.createTopic") }}</button>
       </div>
     </div>
 
-    <div v-if="!tenant || !namespace" class="panel-placeholder">请先选择租户和命名空间</div>
+    <div v-if="!tenant || !namespace" class="panel-placeholder">{{ t("mqTopics.selectTenantNamespace") }}</div>
 
     <div v-else-if="error" class="panel-error">{{ error }}</div>
 
-    <div v-else-if="loading && !topics.length" class="panel-loading">加载中...</div>
+    <div v-else-if="loading && !topics.length" class="panel-loading">{{ t("mqTopics.loading") }}</div>
 
-    <div v-else-if="!topics.length" class="panel-placeholder">该命名空间下暂无主题</div>
+    <div v-else-if="!topics.length" class="panel-placeholder">{{ t("mqTopics.noTopics") }}</div>
+
+    <div v-else-if="!filteredTopics.length" class="panel-placeholder">{{ t("mqTopics.noMatches") }}</div>
 
     <div v-else class="topics-table">
       <table>
         <thead>
           <tr>
-            <th>名称</th>
-            <th>类型</th>
-            <th>分区</th>
-            <th>操作</th>
+            <th>{{ t("mqTopics.name") }}</th>
+            <th>{{ t("mqTopics.type") }}</th>
+            <th>{{ t("mqTopics.partitions") }}</th>
+            <th>{{ t("mqTopics.actions") }}</th>
           </tr>
         </thead>
         <tbody>
@@ -237,21 +272,23 @@ watch(includeNonPersistent, () => {
             <td class="topic-name">
               <div class="topic-name-cell">
                 <span>{{ topic.shortName }}</span>
-                <span v-if="!topic.persistent" class="badge badge-warning">非持久化</span>
+                <span v-if="!topic.persistent" class="badge badge-warning">{{ t("mqTopics.nonPersistent") }}</span>
               </div>
             </td>
             <td>
               <span class="badge" :class="topic.partitioned ? 'badge-info' : 'badge-default'">
-                {{ topic.partitioned ? "分区主题" : "普通主题" }}
+                {{ topic.partitioned ? t("mqTopics.partitionedTopic") : t("mqTopics.normalTopic") }}
               </span>
             </td>
             <td>
-              <span v-if="topic.partitioned">{{ topic.partitions ? `${topic.partitions} 个分区` : "分区数未知" }}</span>
+              <span v-if="topic.partitioned">{{ topic.partitions ? t("mqTopics.partitionCount", { count: topic.partitions }) : t("mqTopics.partitionsUnknown") }}</span>
               <span v-else class="text-muted">-</span>
             </td>
             <td class="actions">
-              <button v-if="topic.partitioned && supportsPartitionedTopics !== false" @click.stop="openPartitionsDialog(topic)" :disabled="readOnly || !topic.partitions" class="btn-sm">调整分区</button>
-              <button @click.stop="handleDelete(topic)" :disabled="readOnly" class="btn-sm btn-danger">删除</button>
+              <button v-if="topic.partitioned && supportsPartitionedTopics !== false" @click.stop="openPartitionsDialog(topic)" :disabled="readOnly || !topic.partitions" class="btn-sm">
+                {{ t("mqTopics.adjustPartitions") }}
+              </button>
+              <button @click.stop="handleDelete(topic)" :disabled="readOnly" class="btn-sm btn-danger">{{ t("mqTopics.delete") }}</button>
             </td>
           </tr>
         </tbody>
@@ -262,41 +299,41 @@ watch(includeNonPersistent, () => {
     <div v-if="showCreateDialog" class="dialog-overlay" @click="showCreateDialog = false">
       <div class="dialog" @click.stop>
         <div class="dialog-header">
-          <h3>创建主题</h3>
+          <h3>{{ t("mqTopics.createTopic") }}</h3>
           <button @click="showCreateDialog = false" class="btn-close">×</button>
         </div>
         <div class="dialog-body">
           <div v-if="!isKafkaCluster" class="form-group">
-            <label>租户 / 命名空间</label>
+            <label>{{ t("mqTopics.tenantNamespace") }}</label>
             <input type="text" :value="`${tenant} / ${namespace}`" disabled />
           </div>
           <div class="form-group">
-            <label>主题名称*</label>
-            <input v-model="formData.topicName" type="text" placeholder="例如: my-topic" :disabled="readOnly" />
+            <label>{{ t("mqTopics.topicName") }}*</label>
+            <input v-model="formData.topicName" type="text" :placeholder="t('mqTopics.topicNamePlaceholder')" :disabled="readOnly" />
           </div>
           <div class="form-group">
             <label class="checkbox-label">
               <input type="checkbox" v-model="formData.persistent" :disabled="readOnly" />
-              持久化主题（推荐）
+              {{ t("mqTopics.persistentRecommended") }}
             </label>
-            <div class="form-hint">持久化主题会将消息保存到磁盘，非持久化主题仅保存在内存中</div>
+            <div class="form-hint">{{ t("mqTopics.persistentHint") }}</div>
           </div>
           <div v-if="supportsPartitionedTopics !== false" class="form-group">
             <label class="checkbox-label">
               <input type="checkbox" v-model="formData.partitioned" :disabled="readOnly" />
-              启用分区
+              {{ t("mqTopics.enablePartitions") }}
             </label>
             <div v-if="formData.partitioned" class="form-subgroup">
-              <label>分区数量*</label>
+              <label>{{ t("mqTopics.partitionQuantity") }}*</label>
               <input v-model.number="formData.partitions" type="number" min="1" max="256" :disabled="readOnly" />
-              <div class="form-hint">分区可以提高并发性能，但会增加资源消耗</div>
+              <div class="form-hint">{{ t("mqTopics.partitionHint") }}</div>
             </div>
           </div>
-          <div v-if="error" class="form-error">{{ error }}</div>
+          <div v-if="dialogError" class="form-error">{{ dialogError }}</div>
         </div>
         <div class="dialog-footer">
-          <button @click="showCreateDialog = false" class="btn-secondary">取消</button>
-          <button @click="handleCreate" :disabled="loading || readOnly" class="btn-primary">创建</button>
+          <button @click="showCreateDialog = false" class="btn-secondary">{{ t("mqTopics.cancel") }}</button>
+          <button @click="handleCreate" :disabled="loading || readOnly" class="btn-primary">{{ t("mqTopics.create") }}</button>
         </div>
       </div>
     </div>
@@ -305,24 +342,24 @@ watch(includeNonPersistent, () => {
     <div v-if="showPartitionsDialog" class="dialog-overlay" @click="showPartitionsDialog = false">
       <div class="dialog" @click.stop>
         <div class="dialog-header">
-          <h3>调整分区数: {{ editingTopic?.shortName }}</h3>
+          <h3>{{ t("mqTopics.updatePartitionsTitle", { name: editingTopic?.shortName }) }}</h3>
           <button @click="showPartitionsDialog = false" class="btn-close">×</button>
         </div>
         <div class="dialog-body">
           <div class="form-group">
-            <label>当前分区数</label>
+            <label>{{ t("mqTopics.currentPartitions") }}</label>
             <input type="number" :value="editingTopic?.partitions" disabled />
           </div>
           <div class="form-group">
-            <label>新分区数*</label>
-            <input v-model.number="newPartitions" type="number" :min="(editingTopic?.partitions || 0) + 1" max="256" :disabled="readOnly" />
-            <div class="form-hint">⚠️ 分区数只能增加，不能减少</div>
+            <label>{{ t("mqTopics.newPartitions") }}*</label>
+            <input v-model.number="newPartitions" type="number" :min="editingCurrentPartitions + 1" max="256" :disabled="readOnly" @change="normalizePartitionInput" @blur="normalizePartitionInput" />
+            <div class="form-hint">{{ t("mqTopics.partitionMinHint", { min: editingCurrentPartitions + 1 }) }}</div>
           </div>
-          <div v-if="error" class="form-error">{{ error }}</div>
+          <div v-if="dialogError" class="form-error">{{ dialogError }}</div>
         </div>
         <div class="dialog-footer">
-          <button @click="showPartitionsDialog = false" class="btn-secondary">取消</button>
-          <button @click="handleUpdatePartitions" :disabled="loading || readOnly" class="btn-primary">更新</button>
+          <button @click="showPartitionsDialog = false" class="btn-secondary">{{ t("mqTopics.cancel") }}</button>
+          <button @click="handleUpdatePartitions" :disabled="loading || !canSubmitPartitionUpdate" class="btn-primary">{{ t("mqTopics.update") }}</button>
         </div>
       </div>
     </div>
@@ -331,6 +368,10 @@ watch(includeNonPersistent, () => {
 
 <style scoped>
 .topics-panel {
+  --topics-surface: var(--card, var(--color-background, #ffffff));
+  --topics-header-bg: color-mix(in srgb, var(--secondary, #f5f5f5) 86%, var(--card, #ffffff));
+  --topics-border: var(--border, var(--color-border, #e5e7eb));
+  --topics-border-light: color-mix(in srgb, var(--topics-border) 68%, transparent);
   height: 100%;
   display: flex;
   flex-direction: column;
@@ -348,6 +389,7 @@ watch(includeNonPersistent, () => {
   display: flex;
   align-items: center;
   gap: 16px;
+  min-width: 0;
 }
 
 .toolbar-actions {
@@ -360,6 +402,30 @@ watch(includeNonPersistent, () => {
   margin: 0;
   font-size: 16px;
   font-weight: 600;
+  flex: 0 0 auto;
+}
+
+.topic-search {
+  width: min(320px, 32vw);
+  min-width: 180px;
+  padding: 6px 10px;
+  border: 1px solid var(--color-border);
+  border-radius: 6px;
+  background: var(--color-background);
+  color: var(--color-text);
+  font-size: 13px;
+}
+
+.topic-search:focus {
+  outline: none;
+  border-color: var(--color-primary);
+  box-shadow: 0 0 0 2px var(--color-primary-alpha);
+}
+
+.topic-count {
+  flex: 0 0 auto;
+  color: var(--color-text-tertiary);
+  font-size: 12px;
 }
 
 .checkbox-label {
@@ -387,35 +453,56 @@ watch(includeNonPersistent, () => {
 }
 
 .topics-table {
+  position: relative;
   flex: 1;
   overflow: auto;
-  background: var(--color-background);
+  background: var(--topics-surface);
+}
+
+.topics-table::before {
+  content: "";
+  position: sticky;
+  top: 0;
+  display: block;
+  height: 38px;
+  margin-bottom: -38px;
+  background: var(--topics-header-bg);
+  z-index: 9;
+  box-shadow:
+    0 1px 0 var(--topics-border),
+    0 2px 8px rgba(0, 0, 0, 0.05);
+  pointer-events: none;
 }
 
 table {
+  position: relative;
   width: 100%;
-  border-collapse: collapse;
+  border-collapse: separate;
+  border-spacing: 0;
 }
 
 thead {
   position: sticky;
   top: 0;
-  background: var(--color-background-secondary);
-  z-index: 1;
+  background: var(--topics-header-bg);
+  z-index: 10;
 }
 
 th {
   position: sticky;
   top: 0;
-  z-index: 2;
+  z-index: 11;
   padding: 10px 12px;
   text-align: left;
   font-weight: 600;
   font-size: 13px;
   color: var(--color-text-secondary);
-  background: var(--color-background-secondary);
-  border-bottom: 1px solid var(--color-border);
-  box-shadow: 0 1px 0 var(--color-border);
+  background: var(--topics-header-bg);
+  border-bottom: 1px solid var(--topics-border);
+  background-clip: padding-box;
+  box-shadow:
+    0 1px 0 var(--topics-border),
+    0 2px 6px rgba(0, 0, 0, 0.04);
 }
 
 tbody tr {
@@ -427,13 +514,22 @@ tbody tr:hover {
   background: var(--color-hover);
 }
 
+tbody tr:hover td {
+  background: var(--color-hover);
+}
+
 tbody tr.selected {
   background: var(--color-primary-alpha);
 }
 
+tbody tr.selected td {
+  background: var(--color-primary-alpha);
+}
+
 td {
   padding: 10px 12px;
-  border-bottom: 1px solid var(--color-border-light);
+  border-bottom: 1px solid var(--topics-border-light);
+  background: var(--topics-surface);
 }
 
 .topic-name-cell {
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index fc5bf438f..cb48d26fa 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -301,6 +301,16 @@ export default {
     nacosTls: "TLS",
     nacosTlsSkipVerify: "Skip certificate verification",
     nacosPageSize: "Page Size",
+    kafkaKerberosPrincipal: "Principal",
+    kafkaKerberosKeytab: "Keytab",
+    kafkaKerberosServiceName: "Service Name",
+    kafkaKerberosKrb5Conf: "krb5.conf",
+    kafkaKerberosPrincipalRequired: "Kafka Kerberos principal is required",
+    kafkaKerberosKeytabRequired: "Kafka Kerberos keytab path is required",
+    kafkaKerberosKeytabPlaceholder: "Path on DBX Agent machine, e.g. /etc/security/keytabs/user.keytab",
+    kafkaKerberosKrb5ConfPlaceholder: "Optional path on DBX Agent machine, e.g. /etc/krb5.conf",
+    kafkaKerberosPathHint: "The keytab and krb5.conf paths are read by DBX Agent and must exist on the machine running DBX Agent; files are not uploaded from this browser.",
+    kafkaKerberosAuthHint: "Uses GSSAPI + keytab login. If the server requires encrypted transport, set Security to SASL_SSL; otherwise Auto or SASL_PLAINTEXT can be used.",
     searchDatabasePlaceholder: "Search database types",
     iconView: "Icon view",
     listView: "List view",
@@ -3438,6 +3448,49 @@ export default {
     exportCancelled: "Export cancelled",
     runInBackground: "Run in background",
   },
+  mqTopics: {
+    title: "Topics",
+    searchPlaceholder: "Search topics",
+    includeNonPersistent: "Include non-persistent topics",
+    refresh: "Refresh",
+    refreshing: "Refreshing...",
+    createTopic: "Create Topic",
+    selectTenantNamespace: "Select a tenant and namespace first",
+    loading: "Loading...",
+    noTopics: "No topics in this namespace",
+    noMatches: "No matching topics",
+    name: "Name",
+    type: "Type",
+    partitions: "Partitions",
+    actions: "Actions",
+    nonPersistent: "Non-persistent",
+    partitionedTopic: "Partitioned topic",
+    normalTopic: "Normal topic",
+    partitionCount: "{count} partitions",
+    partitionsUnknown: "Unknown partitions",
+    adjustPartitions: "Adjust partitions",
+    delete: "Delete",
+    tenantNamespace: "Tenant / Namespace",
+    topicName: "Topic name",
+    topicNamePlaceholder: "e.g. my-topic",
+    persistentRecommended: "Persistent topic (recommended)",
+    persistentHint: "Persistent topics store messages on disk; non-persistent topics only keep them in memory.",
+    enablePartitions: "Enable partitions",
+    partitionQuantity: "Partition count",
+    partitionHint: "Partitions can improve concurrency but increase resource usage.",
+    cancel: "Cancel",
+    create: "Create",
+    updatePartitionsTitle: "Adjust partitions: {name}",
+    currentPartitions: "Current partitions",
+    newPartitions: "New partitions",
+    partitionMinHint: "Partitions can only be increased, not decreased. Minimum: {min}",
+    update: "Update",
+    readOnly: "This connection is read-only and cannot perform write operations.",
+    topicNameRequired: "Topic name is required",
+    currentPartitionsUnknown: "Current partition count is unknown, so partitions cannot be adjusted safely.",
+    partitionMustIncrease: "New partition count must be greater than the current partition count.",
+    confirmDelete: 'Delete topic "{name}"? This action cannot be undone.',
+  },
   nacos: {
     configs: "Configs",
     services: "Services",
diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts
index 43c9f1a81..a2bd5551f 100644
--- a/apps/desktop/src/i18n/locales/es.ts
+++ b/apps/desktop/src/i18n/locales/es.ts
@@ -422,6 +422,16 @@ export default withEnglishFallback({
     colorCustom: "Color personalizado",
     cancelConnecting: "Cancelar conexión",
     connectCancelled: "Conexión cancelada",
+    kafkaKerberosPrincipal: "Principal",
+    kafkaKerberosKeytab: "Keytab",
+    kafkaKerberosServiceName: "Nombre del servicio",
+    kafkaKerberosKrb5Conf: "krb5.conf",
+    kafkaKerberosPrincipalRequired: "Kafka Kerberos principal no puede estar vacío",
+    kafkaKerberosKeytabRequired: "La ruta del keytab de Kafka Kerberos no puede estar vacía",
+    kafkaKerberosKeytabPlaceholder: "Ruta en la máquina donde se ejecuta DBX Agent, por ejemplo /etc/security/keytabs/user.keytab",
+    kafkaKerberosKrb5ConfPlaceholder: "Opcional, ruta en la máquina donde se ejecuta DBX Agent, por ejemplo /etc/krb5.conf",
+    kafkaKerberosPathHint: "Las rutas de keytab y krb5.conf son leídas por DBX Agent y deben existir en la máquina donde se ejecuta DBX Agent; no se subirán archivos desde el navegador actual.",
+    kafkaKerberosAuthHint: "Iniciar sesión con GSSAPI + keytab. Si el servidor requiere transmisión cifrada, configure Security como SASL_SSL; de lo contrario, puede usar Auto o SASL_PLAINTEXT.",
   },
   editor: {
     pressToExecute: "Presiona {mod}+Enter para ejecutar",
@@ -3529,4 +3539,47 @@ export default withEnglishFallback({
     expandAll: "Expandir todo",
     collapseAll: "Contraer todo",
   },
+  mqTopics: {
+    title: "Administración de temas",
+    searchPlaceholder: "Buscar topic",
+    includeNonPersistent: "Incluir temas no persistentes",
+    refresh: "Actualizar",
+    refreshing: "Actualizando...",
+    createTopic: "Crear tema",
+    selectTenantNamespace: "Seleccione primero el tenant y el namespace",
+    loading: "Cargando...",
+    noTopics: "No hay temas en este namespace",
+    noMatches: "No se encontraron temas coincidentes",
+    name: "Nombre",
+    type: "Tipo",
+    partitions: "Particiones",
+    actions: "Acciones",
+    nonPersistent: "No persistente",
+    partitionedTopic: "Tema particionado",
+    normalTopic: "Tema normal",
+    partitionCount: "{count} particiones",
+    partitionsUnknown: "Número de particiones desconocido",
+    adjustPartitions: "Ajustar particiones",
+    delete: "Eliminar",
+    tenantNamespace: "Tenant / Namespace",
+    topicName: "Nombre del tema",
+    topicNamePlaceholder: "Ejemplo: my-topic",
+    persistentRecommended: "Tema persistente (recomendado)",
+    persistentHint: "Los temas persistentes guardan los mensajes en disco, los no persistentes solo en memoria.",
+    enablePartitions: "Habilitar particiones",
+    partitionQuantity: "Cantidad de particiones",
+    partitionHint: "Las particiones pueden mejorar el rendimiento concurrente, pero aumentan el consumo de recursos.",
+    cancel: "Cancelar",
+    create: "Crear",
+    updatePartitionsTitle: "Ajustar número de particiones: {name}",
+    currentPartitions: "Particiones actuales",
+    newPartitions: "Nuevo número de particiones",
+    partitionMinHint: "El número de particiones solo puede aumentar, no disminuir. Mínimo: {min}",
+    update: "Actualizar",
+    readOnly: "La conexión actual es de solo lectura, no se pueden realizar operaciones de escritura.",
+    topicNameRequired: "El nombre del topic no puede estar vacío",
+    currentPartitionsUnknown: "Se desconoce el número actual de particiones, no se puede ajustar de forma segura.",
+    partitionMustIncrease: "El nuevo número de particiones debe ser mayor que el actual.",
+    confirmDelete: '¿Está seguro de que desea eliminar el tema "{name}"? Esta operación no se puede deshacer.',
+  },
 });
diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts
index c5e4c3836..6480822ea 100644
--- a/apps/desktop/src/i18n/locales/it.ts
+++ b/apps/desktop/src/i18n/locales/it.ts
@@ -420,6 +420,16 @@ export default withEnglishFallback({
     colorCustom: "Colore personalizzato",
     cancelConnecting: "Annulla connessione",
     connectCancelled: "Connessione annullata",
+    kafkaKerberosPrincipal: "Principal",
+    kafkaKerberosKeytab: "Keytab",
+    kafkaKerberosServiceName: "Nome del servizio",
+    kafkaKerberosKrb5Conf: "krb5.conf",
+    kafkaKerberosPrincipalRequired: "Kafka Kerberos principal non può essere vuoto",
+    kafkaKerberosKeytabRequired: "Il percorso del keytab Kafka Kerberos non può essere vuoto",
+    kafkaKerberosKeytabPlaceholder: "Percorso sulla macchina DBX Agent, ad esempio /etc/security/keytabs/user.keytab",
+    kafkaKerberosKrb5ConfPlaceholder: "Opzionale, percorso sulla macchina DBX Agent, ad esempio /etc/krb5.conf",
+    kafkaKerberosPathHint: "I percorsi di keytab e krb5.conf vengono letti da DBX Agent e devono esistere sulla macchina in cui è in esecuzione DBX Agent; non vengono caricati file dal browser corrente.",
+    kafkaKerberosAuthHint: "Accedi con GSSAPI + keytab. Se il server richiede trasmissione crittografata, imposta Security su SASL_SSL; altrimenti puoi utilizzare Auto o SASL_PLAINTEXT.",
   },
   editor: {
     pressToExecute: "Premi {mod}+Enter per eseguire",
@@ -3527,4 +3537,47 @@ export default withEnglishFallback({
     expandAll: "Espandi tutto",
     collapseAll: "Comprimi tutto",
   },
+  mqTopics: {
+    title: "Gestione topic",
+    searchPlaceholder: "Cerca topic",
+    includeNonPersistent: "Includi topic non persistenti",
+    refresh: "Aggiorna",
+    refreshing: "Aggiornamento in corso...",
+    createTopic: "Crea topic",
+    selectTenantNamespace: "Seleziona prima tenant e namespace",
+    loading: "Caricamento in corso...",
+    noTopics: "Nessun topic in questo namespace",
+    noMatches: "Nessun topic corrispondente",
+    name: "Nome",
+    type: "Tipo",
+    partitions: "Partizioni",
+    actions: "Azioni",
+    nonPersistent: "Non persistente",
+    partitionedTopic: "Topic partizionato",
+    normalTopic: "Topic normale",
+    partitionCount: "{count} partizioni",
+    partitionsUnknown: "Numero di partizioni sconosciuto",
+    adjustPartitions: "Regola partizioni",
+    delete: "Elimina",
+    tenantNamespace: "Tenant / Namespace",
+    topicName: "Nome topic",
+    topicNamePlaceholder: "es. my-topic",
+    persistentRecommended: "Topic persistente (consigliato)",
+    persistentHint: "I topic persistenti salvano i messaggi su disco, quelli non persistenti solo in memoria.",
+    enablePartitions: "Abilita partizioni",
+    partitionQuantity: "Numero di partizioni",
+    partitionHint: "Le partizioni migliorano le prestazioni di concorrenza ma aumentano il consumo di risorse.",
+    cancel: "Annulla",
+    create: "Crea",
+    updatePartitionsTitle: "Regola numero partizioni: {name}",
+    currentPartitions: "Partizioni attuali",
+    newPartitions: "Nuove partizioni",
+    partitionMinHint: "Il numero di partizioni può solo aumentare, non diminuire. Minimo: {min}",
+    update: "Aggiorna",
+    readOnly: "La connessione corrente è in modalità sola lettura, impossibile eseguire operazioni di scrittura.",
+    topicNameRequired: "Il nome del topic non può essere vuoto",
+    currentPartitionsUnknown: "Numero di partizioni attuali sconosciuto, impossibile regolare le partizioni in modo sicuro.",
+    partitionMustIncrease: "Il nuovo numero di partizioni deve essere maggiore di quello attuale.",
+    confirmDelete: 'Confermi di eliminare il topic "{name}"? Questa operazione è irreversibile.',
+  },
 });
diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts
index 11f52331d..1f75c203f 100644
--- a/apps/desktop/src/i18n/locales/ja.ts
+++ b/apps/desktop/src/i18n/locales/ja.ts
@@ -420,6 +420,16 @@ export default withEnglishFallback({
     zookeeperCreateModeEphemeralSequential: "エフェメラル順序",
     cancelConnecting: "接続をキャンセル",
     connectCancelled: "接続がキャンセルされました",
+    kafkaKerberosPrincipal: "プリンシパル",
+    kafkaKerberosKeytab: "Keytab",
+    kafkaKerberosServiceName: "サービス名",
+    kafkaKerberosKrb5Conf: "krb5.conf",
+    kafkaKerberosPrincipalRequired: "Kafka Kerberos プリンシパルは必須です",
+    kafkaKerberosKeytabRequired: "Kafka Kerberos keytab のパスは必須です",
+    kafkaKerberosKeytabPlaceholder: "DBX Agent が動作するマシンのパス。例: /etc/security/keytabs/user.keytab",
+    kafkaKerberosKrb5ConfPlaceholder: "オプション。DBX Agent が動作するマシンのパス。例: /etc/krb5.conf",
+    kafkaKerberosPathHint: "keytab と krb5.conf のパスは DBX Agent が読み取るため、DBX Agent が動作するマシン上に存在する必要があります。ブラウザからファイルをアップロードすることはありません。",
+    kafkaKerberosAuthHint: "GSSAPI + keytab を使用してログインします。サーバーが暗号化転送を要求する場合は、Security を SASL_SSL に設定してください。それ以外の場合は Auto または SASL_PLAINTEXT を使用できます。",
   },
   editor: {
     pressToExecute: "{mod}+Enter で実行",
@@ -3527,4 +3537,47 @@ export default withEnglishFallback({
     expandAll: "すべて展開",
     collapseAll: "すべて折りたたむ",
   },
+  mqTopics: {
+    title: "トピック管理",
+    searchPlaceholder: "トピックを検索",
+    includeNonPersistent: "非永続トピックを含める",
+    refresh: "更新",
+    refreshing: "更新中...",
+    createTopic: "トピックを作成",
+    selectTenantNamespace: "テナントと名前空間を先に選択してください",
+    loading: "読み込み中...",
+    noTopics: "この名前空間にはトピックがありません",
+    noMatches: "一致するトピックはありません",
+    name: "名前",
+    type: "タイプ",
+    partitions: "パーティション",
+    actions: "操作",
+    nonPersistent: "非永続",
+    partitionedTopic: "パーティショントピック",
+    normalTopic: "通常トピック",
+    partitionCount: "{count} パーティション",
+    partitionsUnknown: "パーティション数不明",
+    adjustPartitions: "パーティションを調整",
+    delete: "削除",
+    tenantNamespace: "テナント / 名前空間",
+    topicName: "トピック名",
+    topicNamePlaceholder: "例: my-topic",
+    persistentRecommended: "永続トピック（推奨）",
+    persistentHint: "永続トピックはメッセージをディスクに保存しますが、非永続トピックはメモリのみに保存されます。",
+    enablePartitions: "パーティションを有効にする",
+    partitionQuantity: "パーティション数",
+    partitionHint: "パーティションは並列性パフォーマンスを向上させますが、リソース消費が増加します。",
+    cancel: "キャンセル",
+    create: "作成",
+    updatePartitionsTitle: "パーティション数の調整: {name}",
+    currentPartitions: "現在のパーティション数",
+    newPartitions: "新しいパーティション数",
+    partitionMinHint: "パーティション数は増やすことのみ可能で、減らすことはできません。最小値: {min}",
+    update: "更新",
+    readOnly: "現在の接続は読み取り専用モードです。書き込み操作は実行できません。",
+    topicNameRequired: "トピック名は必須です",
+    currentPartitionsUnknown: "現在のパーティション数が不明なため、安全にパーティションを調整できません。",
+    partitionMustIncrease: "新しいパーティション数は現在のパーティション数より大きくなければなりません。",
+    confirmDelete: "トピック「{name}」を削除してもよろしいですか？この操作は元に戻せません。",
+  },
 });
diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts
index 9315b9643..e712a33ce 100644
--- a/apps/desktop/src/i18n/locales/pt-BR.ts
+++ b/apps/desktop/src/i18n/locales/pt-BR.ts
@@ -421,6 +421,16 @@ export default withEnglishFallback({
     colorCustom: "Cor personalizada",
     cancelConnecting: "Cancelar conexão",
     connectCancelled: "Conexão cancelada",
+    kafkaKerberosPrincipal: "Principal",
+    kafkaKerberosKeytab: "Keytab",
+    kafkaKerberosServiceName: "Nome do serviço",
+    kafkaKerberosKrb5Conf: "krb5.conf",
+    kafkaKerberosPrincipalRequired: "O principal do Kafka Kerberos não pode estar vazio",
+    kafkaKerberosKeytabRequired: "O caminho do keytab do Kafka Kerberos não pode estar vazio",
+    kafkaKerberosKeytabPlaceholder: "Caminho na máquina do DBX Agent, por exemplo /etc/security/keytabs/user.keytab",
+    kafkaKerberosKrb5ConfPlaceholder: "Opcional, caminho na máquina do DBX Agent, por exemplo /etc/krb5.conf",
+    kafkaKerberosPathHint: "Os caminhos do keytab e do krb5.conf são lidos pelo DBX Agent e devem existir na máquina onde o DBX Agent é executado; os arquivos não são enviados do navegador atual.",
+    kafkaKerberosAuthHint: "Faça login usando GSSAPI + keytab. Se o servidor exigir transmissão criptografada, defina Security como SASL_SSL; caso contrário, você pode usar Auto ou SASL_PLAINTEXT.",
   },
   editor: {
     pressToExecute: "Pressione {mod}+Enter para executar",
@@ -3528,4 +3538,47 @@ export default withEnglishFallback({
     expandAll: "Expandir todos",
     collapseAll: "Recolher todos",
   },
+  mqTopics: {
+    title: "Gerenciamento de tópicos",
+    searchPlaceholder: "Pesquisar tópico",
+    includeNonPersistent: "Incluir tópicos não persistentes",
+    refresh: "Atualizar",
+    refreshing: "Atualizando...",
+    createTopic: "Criar tópico",
+    selectTenantNamespace: "Selecione primeiro o locatário e o namespace",
+    loading: "Carregando...",
+    noTopics: "Nenhum tópico neste namespace",
+    noMatches: "Nenhum tópico correspondente",
+    name: "Nome",
+    type: "Tipo",
+    partitions: "Partições",
+    actions: "Ações",
+    nonPersistent: "Não persistente",
+    partitionedTopic: "Tópico particionado",
+    normalTopic: "Tópico normal",
+    partitionCount: "{count} partições",
+    partitionsUnknown: "Número de partições desconhecido",
+    adjustPartitions: "Ajustar partições",
+    delete: "Excluir",
+    tenantNamespace: "Locatário / Namespace",
+    topicName: "Nome do tópico",
+    topicNamePlaceholder: "Por exemplo: my-topic",
+    persistentRecommended: "Tópico persistente (recomendado)",
+    persistentHint: "Tópicos persistentes salvam as mensagens em disco, tópicos não persistentes salvam apenas na memória.",
+    enablePartitions: "Habilitar partições",
+    partitionQuantity: "Número de partições",
+    partitionHint: "As partições melhoram o desempenho de concorrência, mas aumentam o consumo de recursos.",
+    cancel: "Cancelar",
+    create: "Criar",
+    updatePartitionsTitle: "Ajustar número de partições: {name}",
+    currentPartitions: "Partições atuais",
+    newPartitions: "Novas partições",
+    partitionMinHint: "O número de partições só pode ser aumentado, não reduzido. Mínimo: {min}",
+    update: "Atualizar",
+    readOnly: "A conexão atual é somente leitura, não é possível executar operações de gravação.",
+    topicNameRequired: "O nome do tópico não pode estar vazio",
+    currentPartitionsUnknown: "O número atual de partições é desconhecido, não é possível ajustar com segurança.",
+    partitionMustIncrease: "O novo número de partições deve ser maior que o atual.",
+    confirmDelete: 'Tem certeza de que deseja excluir o tópico "{name}"? Esta operação é irreversível.',
+  },
 });
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index 128d2294c..0038a17a8 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -303,6 +303,16 @@ export default withEnglishFallback({
     nacosTls: "TLS",
     nacosTlsSkipVerify: "跳过证书验证",
     nacosPageSize: "分页大小",
+    kafkaKerberosPrincipal: "Principal",
+    kafkaKerberosKeytab: "Keytab",
+    kafkaKerberosServiceName: "服务名",
+    kafkaKerberosKrb5Conf: "krb5.conf",
+    kafkaKerberosPrincipalRequired: "Kafka Kerberos principal 不能为空",
+    kafkaKerberosKeytabRequired: "Kafka Kerberos keytab 路径不能为空",
+    kafkaKerberosKeytabPlaceholder: "DBX Agent 所在机器路径，例如 /etc/security/keytabs/user.keytab",
+    kafkaKerberosKrb5ConfPlaceholder: "可选，DBX Agent 所在机器路径，例如 /etc/krb5.conf",
+    kafkaKerberosPathHint: "keytab 和 krb5.conf 路径由 DBX Agent 读取，必须存在于运行 DBX Agent 的机器上；不会从当前浏览器上传文件。",
+    kafkaKerberosAuthHint: "使用 GSSAPI + keytab 登录。若服务端要求加密传输，请将 Security 设为 SASL_SSL；否则可使用 Auto 或 SASL_PLAINTEXT。",
     searchDatabasePlaceholder: "搜索数据库类型",
     iconView: "图标视图",
     listView: "列表视图",
@@ -3438,6 +3448,49 @@ export default withEnglishFallback({
     exportCancelled: "导出已取消",
     runInBackground: "后台运行",
   },
+  mqTopics: {
+    title: "主题管理",
+    searchPlaceholder: "搜索 topic",
+    includeNonPersistent: "包含非持久化主题",
+    refresh: "刷新",
+    refreshing: "刷新中...",
+    createTopic: "创建主题",
+    selectTenantNamespace: "请先选择租户和命名空间",
+    loading: "加载中...",
+    noTopics: "该命名空间下暂无主题",
+    noMatches: "没有匹配的主题",
+    name: "名称",
+    type: "类型",
+    partitions: "分区",
+    actions: "操作",
+    nonPersistent: "非持久化",
+    partitionedTopic: "分区主题",
+    normalTopic: "普通主题",
+    partitionCount: "{count} 个分区",
+    partitionsUnknown: "分区数未知",
+    adjustPartitions: "调整分区",
+    delete: "删除",
+    tenantNamespace: "租户 / 命名空间",
+    topicName: "主题名称",
+    topicNamePlaceholder: "例如: my-topic",
+    persistentRecommended: "持久化主题（推荐）",
+    persistentHint: "持久化主题会将消息保存到磁盘，非持久化主题仅保存在内存中。",
+    enablePartitions: "启用分区",
+    partitionQuantity: "分区数量",
+    partitionHint: "分区可以提高并发性能，但会增加资源消耗。",
+    cancel: "取消",
+    create: "创建",
+    updatePartitionsTitle: "调整分区数: {name}",
+    currentPartitions: "当前分区数",
+    newPartitions: "新分区数",
+    partitionMinHint: "分区数只能增加，不能减少。最小值：{min}",
+    update: "更新",
+    readOnly: "当前连接为只读模式，不能执行写操作。",
+    topicNameRequired: "Topic name 不能为空",
+    currentPartitionsUnknown: "当前分区数未知，无法安全调整分区。",
+    partitionMustIncrease: "新分区数必须大于当前分区数。",
+    confirmDelete: "确定要删除主题「{name}」吗？此操作不可逆。",
+  },
   nacos: {
     configs: "配置",
     services: "服务",
diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts
index d44c192fc..921811ed4 100644
--- a/apps/desktop/src/i18n/locales/zh-TW.ts
+++ b/apps/desktop/src/i18n/locales/zh-TW.ts
@@ -421,6 +421,16 @@ export default withEnglishFallback({
     colorCustom: "自訂顏色",
     cancelConnecting: "取消連接",
     connectCancelled: "連接已取消",
+    kafkaKerberosPrincipal: "Principal",
+    kafkaKerberosKeytab: "Keytab",
+    kafkaKerberosServiceName: "服務名稱",
+    kafkaKerberosKrb5Conf: "krb5.conf",
+    kafkaKerberosPrincipalRequired: "Kafka Kerberos principal 不能為空",
+    kafkaKerberosKeytabRequired: "Kafka Kerberos keytab 路徑不能為空",
+    kafkaKerberosKeytabPlaceholder: "DBX Agent 所在機器路徑，例如 /etc/security/keytabs/user.keytab",
+    kafkaKerberosKrb5ConfPlaceholder: "可選，DBX Agent 所在機器路徑，例如 /etc/krb5.conf",
+    kafkaKerberosPathHint: "keytab 和 krb5.conf 路徑由 DBX Agent 讀取，必須存在於執行 DBX Agent 的機器上；不會從當前瀏覽器上傳檔案。",
+    kafkaKerberosAuthHint: "使用 GSSAPI + keytab 登入。若伺服器端要求加密傳輸，請將 Security 設為 SASL_SSL；否則可使用 Auto 或 SASL_PLAINTEXT。",
   },
   editor: {
     pressToExecute: "按 {mod}+Enter 執行查詢",
@@ -3528,4 +3538,47 @@ export default withEnglishFallback({
     expandAll: "全部展開",
     collapseAll: "全部摺疊",
   },
+  mqTopics: {
+    title: "主題管理",
+    searchPlaceholder: "搜尋 topic",
+    includeNonPersistent: "包含非持久化主題",
+    refresh: "重新整理",
+    refreshing: "重新整理中...",
+    createTopic: "建立主題",
+    selectTenantNamespace: "請先選取租用戶和命名空間",
+    loading: "載入中...",
+    noTopics: "該命名空間下暫無主題",
+    noMatches: "沒有符合的主題",
+    name: "名稱",
+    type: "類型",
+    partitions: "分區",
+    actions: "操作",
+    nonPersistent: "非持久化",
+    partitionedTopic: "分區主題",
+    normalTopic: "一般主題",
+    partitionCount: "{count} 個分區",
+    partitionsUnknown: "分區數未知",
+    adjustPartitions: "調整分區",
+    delete: "刪除",
+    tenantNamespace: "租用戶 / 命名空間",
+    topicName: "主題名稱",
+    topicNamePlaceholder: "例如: my-topic",
+    persistentRecommended: "持久化主題（推薦）",
+    persistentHint: "持久化主題會將訊息儲存到磁碟，非持久化主題僅儲存在記憶體中。",
+    enablePartitions: "啟用分區",
+    partitionQuantity: "分區數量",
+    partitionHint: "分區可以提高並行效能，但會增加資源消耗。",
+    cancel: "取消",
+    create: "建立",
+    updatePartitionsTitle: "調整分區數: {name}",
+    currentPartitions: "目前分區數",
+    newPartitions: "新分區數",
+    partitionMinHint: "分區數只能增加，不能減少。最小值：{min}",
+    update: "更新",
+    readOnly: "目前連線為唯讀模式，無法執行寫入操作。",
+    topicNameRequired: "Topic name 不能為空",
+    currentPartitionsUnknown: "目前分區數未知，無法安全調整分區。",
+    partitionMustIncrease: "新分區數必須大於目前分區數。",
+    confirmDelete: "確定要刪除主題「{name}」嗎？此操作無法復原。",
+  },
 });
diff --git a/apps/desktop/src/lib/connection/mqAuth.ts b/apps/desktop/src/lib/connection/mqAuth.ts
new file mode 100644
index 000000000..a6611be9c
--- /dev/null
+++ b/apps/desktop/src/lib/connection/mqAuth.ts
@@ -0,0 +1,22 @@
+import type { MqAuth, MqSystemKind } from "@/types/mq";
+
+export type MqAuthKind = MqAuth["kind"];
+export type MqUiAuthKind = MqAuthKind | "kerberos";
+
+const KAFKA_AUTH_KINDS = new Set<MqUiAuthKind>(["none", "basic", "kerberos"]);
+
+export function isMqAuthKindAllowedForSystem(systemKind: MqSystemKind, authKind: MqUiAuthKind): boolean {
+  if (systemKind === "kafka") return KAFKA_AUTH_KINDS.has(authKind);
+  return authKind !== "kerberos";
+}
+
+export function detectMqUiAuthKind({ systemKind, authKind, saslMechanism, jaasConfig }: { systemKind: MqSystemKind; authKind?: MqAuthKind; saslMechanism: string; jaasConfig: string }): MqUiAuthKind {
+  if (systemKind === "kafka") {
+    if (saslMechanism.toUpperCase() === "GSSAPI" && jaasConfig.includes("Krb5LoginModule")) {
+      return "kerberos";
+    }
+    return authKind === "basic" ? "basic" : "none";
+  }
+
+  return authKind || "none";
+}
diff --git a/crates/dbx-core/src/mq/adapters/kafka.rs b/crates/dbx-core/src/mq/adapters/kafka.rs
index 8fe25f7d0..9dabbe689 100644
--- a/crates/dbx-core/src/mq/adapters/kafka.rs
+++ b/crates/dbx-core/src/mq/adapters/kafka.rs
@@ -789,6 +789,34 @@ mod tests {
         assert_eq!(params.pointer("/properties/client.id").and_then(|v| v.as_str()), Some("dbx"));
     }
 
+    #[test]
+    fn connection_params_preserve_kafka_gssapi_properties() {
+        let cfg = kafka_config(
+            serde_json::json!({
+                "bootstrapServers": "broker:9093",
+                "securityProtocol": "SASL_SSL",
+                "saslMechanism": "GSSAPI",
+                "properties": {
+                    "sasl.jaas.config": "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab=\"/tmp/user.keytab\" principal=\"user@EXAMPLE.COM\";",
+                    "sasl.kerberos.service.name": "kafka",
+                    "java.security.krb5.conf": "/tmp/krb5.conf"
+                }
+            }),
+            MqAuth::None,
+            false,
+        );
+
+        let params = build_connection_params(&cfg);
+
+        assert_eq!(params.get("security_protocol").and_then(|v| v.as_str()), Some("SASL_SSL"));
+        assert_eq!(params.get("sasl_mechanism").and_then(|v| v.as_str()), Some("GSSAPI"));
+        assert_eq!(params.pointer("/properties/sasl.kerberos.service.name").and_then(|v| v.as_str()), Some("kafka"));
+        assert_eq!(
+            params.pointer("/properties/java.security.krb5.conf").and_then(|v| v.as_str()),
+            Some("/tmp/krb5.conf")
+        );
+    }
+
     #[test]
     fn reset_cursor_params_preserve_timestamp_position() {
         let topic = TopicRef {
diff --git a/packages/app-tests/mqAuth.test.ts b/packages/app-tests/mqAuth.test.ts
new file mode 100644
index 000000000..df4d0b6d3
--- /dev/null
+++ b/packages/app-tests/mqAuth.test.ts
@@ -0,0 +1,32 @@
+import assert from "node:assert/strict";
+import { test } from "vitest";
+
+import { detectMqUiAuthKind, isMqAuthKindAllowedForSystem } from "../../apps/desktop/src/lib/connection/mqAuth.ts";
+
+test("keeps hydrated Kafka Kerberos auth allowed during edit", () => {
+  const authKind = detectMqUiAuthKind({
+    systemKind: "kafka",
+    authKind: "none",
+    saslMechanism: "GSSAPI",
+    jaasConfig: 'com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="/etc/user.keytab" principal="user@EXAMPLE.COM";',
+  });
+
+  assert.equal(authKind, "kerberos");
+  assert.equal(isMqAuthKindAllowedForSystem("kafka", authKind), true);
+});
+
+test("normalizes unsupported Kafka auth kinds to none", () => {
+  assert.equal(
+    detectMqUiAuthKind({
+      systemKind: "kafka",
+      authKind: "token",
+      saslMechanism: "PLAIN",
+      jaasConfig: "",
+    }),
+    "none",
+  );
+});
+
+test("does not allow Kerberos auth outside Kafka", () => {
+  assert.equal(isMqAuthKindAllowedForSystem("pulsar", "kerberos"), false);
+});