✨ feat(kafka): 新增 Kafka 数据源连接支持

Refs #387
2026-06-18 20:49:45 +08:00 · 2026-06-13 21:11:08 +08:00
parent d2f68acae8
commit 0ff17dc27c
37 changed files with 1989 additions and 12 deletions
--- a/frontend/src/components/ConnectionModal.edit-password.test.tsx
+++ b/frontend/src/components/ConnectionModal.edit-password.test.tsx
@@ -33,10 +33,10 @@ describe('ConnectionModal data source registry', () => {
    expect(source).toContain('type === "elasticsearch"');
    expect(source).toContain("return '支持索引浏览、Mapping 检查、JSON DSL 和 query_string 查询';");
    expect(source).toContain(
-      'type === "clickhouse" ? "default" : (type === "redis" || type === "elasticsearch" || type === "chroma" || type === "qdrant") ? "" : "root";',
+      'type === "clickhouse" ? "default" : (type === "redis" || type === "elasticsearch" || type === "chroma" || type === "qdrant" || type === "kafka") ? "" : "root";',
    );
    expect(source).toContain(
-      'placeholder={(dbType === "elasticsearch" || dbType === "chroma" || dbType === "qdrant") ? "未开启认证可留空" : undefined}',
+      'placeholder={(dbType === "elasticsearch" || dbType === "chroma" || dbType === "qdrant" || dbType === "kafka") ? "未开启认证可留空" : undefined}',
    );
    expect(source).toContain('label="显示数据库 (留空显示全部)"');
  });
@@ -77,6 +77,18 @@ describe('ConnectionModal data source registry', () => {
    expect(source).toContain('return "fetchSize=1024&timeZone=Asia%2FShanghai";');
  });

+  it('exposes Kafka in the create-connection picker with broker and topic defaults', () => {
+    expect(source).toContain("case 'kafka':");
+    expect(source).toContain('return 9092;');
+    expect(source).toContain("key: 'kafka'");
+    expect(source).toContain("name: 'Kafka'");
+    expect(source).toContain('dbType === "kafka"');
+    expect(source).toContain("return 'Broker / Topic / Consumer Group';");
+    expect(source).toContain('return "kafka://user:pass@127.0.0.1:9092,127.0.0.2:9092/orders.events?topology=cluster&groupId=analytics&mechanism=scram-sha-256";');
+    expect(source).toContain('return "groupId=gonavi&mechanism=scram-sha-256&clientId=gonavi-desktop&startOffset=latest";');
+    expect(source).toContain('label="默认 Topic（可选）"');
+  });
+
  it('exposes GaussDB in the create-connection picker with PostgreSQL-family defaults', () => {
    expect(source).toContain("case 'gaussdb':");
    expect(source).toContain('return 5432;');
--- a/frontend/src/components/ConnectionModal.tsx
+++ b/frontend/src/components/ConnectionModal.tsx
@@ -385,6 +385,7 @@ const ConnectionModal: React.FC<{
  );
  const disableLocalBackdropFilter = isMacLikePlatform();
  const mysqlTopology = Form.useWatch("mysqlTopology", form) || "single";
+  const kafkaTopology = Form.useWatch("kafkaTopology", form) || "single";
  const mongoTopology = Form.useWatch("mongoTopology", form) || "single";
  const mongoSrv = Form.useWatch("mongoSrv", form) || false;
  const redisTopology = Form.useWatch("redisTopology", form) || "single";
@@ -418,6 +419,7 @@ const ConnectionModal: React.FC<{
  );
  const isOceanBaseOracle = dbType === "oceanbase" && oceanBaseProtocol === "oracle";
  const isMySQLLike = isMySQLCompatibleType(dbType) && !isOceanBaseOracle;
+  const isKafka = dbType === "kafka";
  const supportsConnectionParams = supportsConnectionParamsForType(dbType);
  const isSSLType = supportsSSLForType(dbType);
  const supportsSSLCAPath = supportsSSLCAPathForType(dbType);
@@ -1629,6 +1631,62 @@ const ConnectionModal: React.FC<{
      };
    }

+    if (type === "kafka") {
+      const defaultPort = getDefaultPortByType(type);
+      const parsed =
+        parseMultiHostUri(trimmedUri, "kafka") ||
+        parseMultiHostUri(trimmedUri, "apache-kafka") ||
+        parseMultiHostUri(trimmedUri, "apache_kafka");
+      if (!parsed) {
+        return null;
+      }
+      if (!parsed.hosts.length || parsed.hosts.length > MAX_URI_HOSTS) {
+        return null;
+      }
+      if (parsed.hosts.some((entry) => !isValidUriHostEntry(entry))) {
+        return null;
+      }
+      const hostList = normalizeAddressList(parsed.hosts, defaultPort);
+      if (!hostList.length) {
+        return null;
+      }
+      const primary = parseHostPort(
+        hostList[0] || `localhost:${defaultPort}`,
+        defaultPort,
+      );
+      const tlsEnabled = normalizeUriBool(
+        parsed.params.get("tls") ||
+          parsed.params.get("ssl") ||
+          parsed.params.get("useSSL") ||
+          parsed.params.get("use_ssl"),
+      );
+      const skipVerify = normalizeUriBool(
+        parsed.params.get("skip_verify") || parsed.params.get("skipVerify"),
+      );
+      const topology = String(parsed.params.get("topology") || "")
+        .trim()
+        .toLowerCase();
+      const timeoutValue = Number(parsed.params.get("timeout"));
+      return {
+        host: primary?.host || "localhost",
+        port: primary?.port || defaultPort,
+        user: parsed.username,
+        password: parsed.password,
+        database: parsed.database || "",
+        useSSL: tlsEnabled,
+        sslMode: tlsEnabled ? (skipVerify ? "skip-verify" : "required") : "disable",
+        ...extractSSLPathValuesFromParams(parsed.params, type),
+        kafkaTopology:
+          topology === "cluster" || hostList.length > 1 ? "cluster" : "single",
+        kafkaHosts: hostList.slice(1),
+        connectionParams: serializeConnectionParams(parsed.params),
+        timeout:
+          Number.isFinite(timeoutValue) && timeoutValue > 0
+            ? Math.min(MAX_TIMEOUT_SECONDS, Math.trunc(timeoutValue))
+            : undefined,
+      };
+    }
+
    if (type === "clickhouse") {
      const httpValues = parseClickHouseHTTPUriToValues(trimmedUri);
      if (httpValues) {
@@ -1877,6 +1935,9 @@ const ConnectionModal: React.FC<{
    if (dbType === "iotdb") {
      return "iotdb://root:root@127.0.0.1:6667/root.sg";
    }
+    if (dbType === "kafka") {
+      return "kafka://user:pass@127.0.0.1:9092,127.0.0.2:9092/orders.events?topology=cluster&groupId=analytics&mechanism=scram-sha-256";
+    }
    if (dbType === "redis") {
      return "redis://:pass@127.0.0.1:6379,127.0.0.2:6379/0?topology=cluster 或 redis://:pass@10.0.0.1:26379,10.0.0.2:26379/0?topology=sentinel&master=mymaster";
    }
@@ -1932,6 +1993,8 @@ const ConnectionModal: React.FC<{
        return "timezone=Asia%2FShanghai";
      case "iotdb":
        return "fetchSize=1024&timeZone=Asia%2FShanghai";
+      case "kafka":
+        return "groupId=gonavi&mechanism=scram-sha-256&clientId=gonavi-desktop&startOffset=latest";
      default:
        return "key=value&another=value";
    }
@@ -1994,6 +2057,36 @@ const ConnectionModal: React.FC<{
      return `${scheme}://${encodedAuth}${hosts.join(",")}${dbPath}${query ? `?${query}` : ""}`;
    }

+    if (type === "kafka") {
+      const primary = toAddress(host, port, defaultPort);
+      const brokers =
+        values.kafkaTopology === "cluster"
+          ? normalizeAddressList(values.kafkaHosts, defaultPort)
+          : [];
+      const allBrokers = normalizeAddressList([primary, ...brokers], defaultPort);
+      const params = new URLSearchParams();
+      if (allBrokers.length > 1 || values.kafkaTopology === "cluster") {
+        params.set("topology", "cluster");
+      }
+      if (values.useSSL) {
+        const mode = String(values.sslMode || "preferred")
+          .trim()
+          .toLowerCase();
+        params.set("tls", "true");
+        if (mode === "skip-verify" || mode === "preferred") {
+          params.set("skip_verify", "true");
+        }
+        appendSSLPathParamsForUri(params, type, values);
+      }
+      if (Number.isFinite(timeout) && timeout > 0) {
+        params.set("timeout", String(timeout));
+      }
+      mergeConnectionParams(params, values.connectionParams);
+      const topicPath = database ? `/${encodeURIComponent(database)}` : "";
+      const query = params.toString();
+      return `kafka://${encodedAuth}${allBrokers.join(",")}${topicPath}${query ? `?${query}` : ""}`;
+    }
+
    if (type === "redis") {
      return buildRedisUriFromValues(values);
    }
@@ -2364,6 +2457,8 @@ const ConnectionModal: React.FC<{
          configType === "sphinx"
            ? normalizedHosts.slice(1)
            : [];
+        const kafkaHosts =
+          configType === "kafka" ? normalizedHosts.slice(1) : [];
        const mongoHosts =
          configType === "mongodb" ? normalizedHosts.slice(1) : [];
        const redisHosts =
@@ -2371,6 +2466,9 @@ const ConnectionModal: React.FC<{
        const mysqlIsReplica =
          String(config.topology || "").toLowerCase() === "replica" ||
          mysqlReplicaHosts.length > 0;
+        const kafkaIsCluster =
+          String(config.topology || "").toLowerCase() === "cluster" ||
+          kafkaHosts.length > 0;
        const mongoIsReplica =
          String(config.topology || "").toLowerCase() === "replica" ||
          mongoHosts.length > 0 ||
@@ -2443,6 +2541,8 @@ const ConnectionModal: React.FC<{
          timeout: resolvedJvmTimeout,
          mysqlTopology: mysqlIsReplica ? "replica" : "single",
          mysqlReplicaHosts: mysqlReplicaHosts,
+          kafkaTopology: kafkaIsCluster ? "cluster" : "single",
+          kafkaHosts: kafkaHosts,
          mysqlReplicaUser: config.mysqlReplicaUser || "",
          mysqlReplicaPassword: config.mysqlReplicaPassword || "",
          mongoTopology: mongoIsReplica ? "replica" : "single",
@@ -3401,6 +3501,23 @@ const ConnectionModal: React.FC<{
      }
    }

+    if (type === "kafka") {
+      const brokers =
+        mergedValues.kafkaTopology === "cluster"
+          ? normalizeAddressList(mergedValues.kafkaHosts, defaultPort)
+          : [];
+      const allHosts = normalizeAddressList(
+        [`${primaryHost}:${primaryPort}`, ...brokers],
+        defaultPort,
+      );
+      if (mergedValues.kafkaTopology === "cluster" || allHosts.length > 1) {
+        hosts = allHosts;
+        topology = "cluster";
+      } else {
+        topology = "single";
+      }
+    }
+
    if (type === "mongodb") {
      mongoSrvEnabled = !!mergedValues.mongoSrv;
      const extraHosts =
@@ -3637,6 +3754,7 @@ const ConnectionModal: React.FC<{
        includeDatabases: undefined,
        includeRedisDatabases: undefined,
        mysqlTopology: "single",
+        kafkaTopology: "single",
        redisTopology: "single",
        mongoTopology: "single",
        mongoSrv: false,
@@ -3646,6 +3764,7 @@ const ConnectionModal: React.FC<{
        mongoAuthMechanism: "",
        savePassword: true,
        mysqlReplicaHosts: [],
+        kafkaHosts: [],
        redisHosts: [],
        redisSentinelMaster: "",
        redisSentinelUser: "",
@@ -3699,6 +3818,7 @@ const ConnectionModal: React.FC<{
        httpTunnelUser: "",
        httpTunnelPassword: "",
        mysqlTopology: "single",
+        kafkaTopology: "single",
        redisTopology: "single",
        mongoTopology: "single",
        mongoSrv: false,
@@ -3708,6 +3828,7 @@ const ConnectionModal: React.FC<{
        mongoAuthMechanism: "",
        savePassword: true,
        mysqlReplicaHosts: [],
+        kafkaHosts: [],
        redisHosts: [],
        redisSentinelMaster: "",
        redisSentinelUser: "",
@@ -3722,7 +3843,7 @@ const ConnectionModal: React.FC<{
      });
    } else if (type !== "custom") {
      const defaultUser =
-        type === "clickhouse" ? "default" : (type === "redis" || type === "elasticsearch" || type === "chroma" || type === "qdrant") ? "" : "root";
+        type === "clickhouse" ? "default" : (type === "redis" || type === "elasticsearch" || type === "chroma" || type === "qdrant" || type === "kafka") ? "" : "root";
      const sslCapableType = supportsSSLForType(type);
      setUseSSL(false);
      setUseHttpTunnel(false);
@@ -3741,6 +3862,7 @@ const ConnectionModal: React.FC<{
        httpTunnelUser: "",
        httpTunnelPassword: "",
        mysqlTopology: "single",
+        kafkaTopology: "single",
        redisTopology: "single",
        mongoTopology: "single",
        mongoSrv: false,
@@ -3750,6 +3872,7 @@ const ConnectionModal: React.FC<{
        mongoAuthMechanism: "",
        savePassword: true,
        mysqlReplicaHosts: [],
+        kafkaHosts: [],
        redisHosts: [],
        redisSentinelMaster: "",
        redisSentinelUser: "",
@@ -4850,6 +4973,22 @@ const ConnectionModal: React.FC<{
                  ),
                })}

+              {dbType === "kafka" &&
+                renderConfigSectionCard({
+                  sectionKey: "service",
+                  icon: <DatabaseOutlined />,
+                  children: (
+                    <Form.Item
+                      name="database"
+                      label="默认 Topic（可选）"
+                      help="留空时必须在 SQL 中显式指定 Topic；填写后可直接执行 SHOW、CONSUME 或 SELECT 预览。"
+                      style={{ marginBottom: 0 }}
+                    >
+                      <Input {...noAutoCapInputProps} placeholder="例如：orders.events" />
+                    </Form.Item>
+                  ),
+                })}
+
              {(dbType === "oracle" || isOceanBaseOracle) &&
                renderConfigSectionCard({
                  sectionKey: "service",
@@ -4902,6 +5041,48 @@ const ConnectionModal: React.FC<{
                  }),
                })}

+              {isKafka &&
+                renderConfigSectionCard({
+                  sectionKey: "connectionMode",
+                  icon: <ClusterOutlined />,
+                  children: renderChoiceCards({
+                    fieldName: "kafkaTopology",
+                    value: String(kafkaTopology),
+                    options: [
+                      {
+                        value: "single",
+                        label: "单 Broker",
+                        description: "只配置一个 bootstrap broker，适合本地或简单环境。",
+                      },
+                      {
+                        value: "cluster",
+                        label: "集群模式",
+                        description: "配置多个 bootstrap broker，提高发现与故障切换成功率。",
+                      },
+                    ],
+                  }),
+                })}
+
+              {isKafka &&
+                kafkaTopology === "cluster" &&
+                renderConfigSectionCard({
+                  sectionKey: "replica",
+                  icon: <ClusterOutlined />,
+                  children: (
+                    <Form.Item
+                      name="kafkaHosts"
+                      label="额外 Broker 地址"
+                      help="可输入多个 broker 地址，格式：host:port（回车确认）"
+                    >
+                      <Select
+                        mode="tags"
+                        placeholder="例如：10.10.0.12:9092、10.10.0.13:9092"
+                        tokenSeparators={[",", ";", " "]}
+                      />
+                    </Form.Item>
+                  ),
+                })}
+
              {isMySQLLike &&
                mysqlTopology === "replica" &&
                renderConfigSectionCard({
@@ -5019,13 +5200,13 @@ const ConnectionModal: React.FC<{
                          name="user"
                          label="用户名"
                          rules={
-                            (dbType === "mongodb" || dbType === "elasticsearch" || dbType === "chroma" || dbType === "qdrant")
+                            (dbType === "mongodb" || dbType === "elasticsearch" || dbType === "chroma" || dbType === "qdrant" || dbType === "kafka")
                              ? []
                              : [createUriAwareRequiredRule("请输入用户名")]
                          }
                          style={{ marginBottom: 0 }}
                        >
-                          <Input {...noAutoCapInputProps} placeholder={(dbType === "elasticsearch" || dbType === "chroma" || dbType === "qdrant") ? "未开启认证可留空" : undefined} />
+                          <Input {...noAutoCapInputProps} placeholder={(dbType === "elasticsearch" || dbType === "chroma" || dbType === "qdrant" || dbType === "kafka") ? "未开启认证可留空" : undefined} />
                        </Form.Item>
                        <Form.Item
                          name="password"
@@ -5099,6 +5280,7 @@ const ConnectionModal: React.FC<{

              {!isFileDb &&
                !isRedis &&
+                !isKafka &&
                renderConfigSectionCard({
                  sectionKey: "databaseScope",
                  icon: <DatabaseOutlined />,
@@ -5946,6 +6128,7 @@ const ConnectionModal: React.FC<{
          connectionParams: "",
          oceanBaseProtocol: "mysql",
          mysqlTopology: "single",
+          kafkaTopology: "single",
          redisTopology: "single",
          mongoTopology: "single",
          mongoSrv: false,
@@ -5953,6 +6136,7 @@ const ConnectionModal: React.FC<{
          mongoAuthMechanism: "",
          savePassword: true,
          mysqlReplicaHosts: [],
+          kafkaHosts: [],
          redisHosts: [],
          redisSentinelMaster: "",
          redisSentinelUser: "",
--- a/frontend/src/components/DatabaseIcons.test.tsx
+++ b/frontend/src/components/DatabaseIcons.test.tsx
@@ -39,6 +39,13 @@ describe('DatabaseIcons', () => {
    expect(markup).toContain('>Io</text>');
  });

+  it('includes Kafka in the selectable database icons', () => {
+    expect(DB_ICON_TYPES).toContain('kafka');
+    expect(getDbIconLabel('kafka')).toBe('Kafka');
+    const markup = renderToStaticMarkup(<>{getDbIcon('kafka', undefined, 22)}</>);
+    expect(markup).toContain('>Kf</text>');
+  });
+
  it('includes GaussDB in the selectable database icons', () => {
    expect(DB_ICON_TYPES).toContain('gaussdb');
    expect(getDbIconLabel('gaussdb')).toBe('GaussDB');
--- a/frontend/src/components/DatabaseIcons.tsx
+++ b/frontend/src/components/DatabaseIcons.tsx
@@ -51,6 +51,7 @@ const DB_DEFAULT_COLORS: Record<string, string> = {
    iris:       '#1F6FEB',
    tdengine:   '#2962FF',
    iotdb:      '#0F766E',
+    kafka:      '#F97316',
    chroma:     '#7C3AED',
    qdrant:     '#DC244C',
    diros:      '#0050B3',
@@ -188,6 +189,9 @@ const TDengineIcon: React.FC<DbIconProps> = ({ size = 16, color }) => (
 const IoTDBIcon: React.FC<DbIconProps> = ({ size = 16, color }) => (
    <ColorBadge size={size} color={color || DB_DEFAULT_COLORS.iotdb} label="Io" />
 );
+const KafkaIcon: React.FC<DbIconProps> = ({ size = 16, color }) => (
+    <ColorBadge size={size} color={color || DB_DEFAULT_COLORS.kafka} label="Kf" />
+);
 const ChromaIcon: React.FC<DbIconProps> = ({ size = 16, color }) => (
    <ColorBadge size={size} color={color || DB_DEFAULT_COLORS.chroma} label="Ch" />
 );
@@ -249,6 +253,7 @@ const DB_ICON_MAP: Record<string, React.FC<DbIconProps>> = {
    iris: IrisIcon,
    tdengine: TDengineIcon,
    iotdb: IoTDBIcon,
+    kafka: KafkaIcon,
    chroma: ChromaIcon,
    qdrant: QdrantIcon,
    elasticsearch: ElasticsearchIcon,
@@ -259,7 +264,7 @@ const DB_ICON_MAP: Record<string, React.FC<DbIconProps>> = {
 export const DB_ICON_TYPES: string[] = [
    'mysql', 'mariadb', 'oceanbase', 'postgres', 'redis', 'mongodb', 'jvm',
    'oracle', 'sqlserver', 'sqlite', 'duckdb', 'clickhouse', 'starrocks',
-    'kingbase', 'dameng', 'vastbase', 'opengauss', 'gaussdb', 'highgo', 'iris', 'tdengine', 'iotdb', 'chroma', 'qdrant', 'elasticsearch', 'custom',
+    'kingbase', 'dameng', 'vastbase', 'opengauss', 'gaussdb', 'highgo', 'iris', 'tdengine', 'iotdb', 'kafka', 'chroma', 'qdrant', 'elasticsearch', 'custom',
 ];

 /** 该类型是否有品牌 SVG 文件 */
@@ -281,7 +286,7 @@ export const getDbIconLabel = (type: string): string => {
        sqlserver: 'SQL Server', clickhouse: 'ClickHouse', sqlite: 'SQLite',
        starrocks: 'StarRocks',
        duckdb: 'DuckDB', kingbase: '金仓', dameng: '达梦',
-        vastbase: 'VastBase', opengauss: 'OpenGauss', gaussdb: 'GaussDB', highgo: '瀚高', iris: 'InterSystems IRIS', tdengine: 'TDengine', iotdb: 'Apache IoTDB',
+        vastbase: 'VastBase', opengauss: 'OpenGauss', gaussdb: 'GaussDB', highgo: '瀚高', iris: 'InterSystems IRIS', tdengine: 'TDengine', iotdb: 'Apache IoTDB', kafka: 'Kafka',
        chroma: 'Chroma',
        qdrant: 'Qdrant',
        elasticsearch: 'Elasticsearch',
--- a/frontend/src/store.ts
+++ b/frontend/src/store.ts
@@ -290,6 +290,7 @@ const SUPPORTED_CONNECTION_TYPES = new Set([
  "redis",
  "tdengine",
  "iotdb",
+  "kafka",
  "oracle",
  "dameng",
  "kingbase",
@@ -327,6 +328,7 @@ const SSL_SUPPORTED_CONNECTION_TYPES = new Set([
  "redis",
  "elasticsearch",
  "tdengine",
+  "kafka",
 ]);

 const getDefaultPortByType = (type: string): number => {
@@ -359,6 +361,8 @@ const getDefaultPortByType = (type: string): number => {
      return 6041;
    case "iotdb":
      return 6667;
+    case "kafka":
+      return 9092;
    case "oracle":
      return 1521;
    case "dameng":
@@ -525,6 +529,9 @@ const normalizeConnectionType = (value: unknown): string => {
  if (type === "gaussdb" || type === "gauss_db" || type === "gauss-db") {
    return "gaussdb";
  }
+  if (type === "kafka" || type === "apache-kafka" || type === "apache_kafka") {
+    return "kafka";
+  }
  if (
    type === "inter-systems" ||
    type === "inter-systems-iris" ||
--- a/frontend/src/utils/connectionDriverType.test.ts
+++ b/frontend/src/utils/connectionDriverType.test.ts
@@ -18,6 +18,8 @@ describe('connectionDriverType', () => {
    expect(normalizeDriverType('qdrant-db')).toBe('qdrant');
    expect(normalizeDriverType('apache-iotdb')).toBe('iotdb');
    expect(normalizeDriverType('apache_iotdb')).toBe('iotdb');
+    expect(normalizeDriverType('apache-kafka')).toBe('kafka');
+    expect(normalizeDriverType('apache_kafka')).toBe('kafka');
    expect(normalizeDriverType('doris')).toBe('diros');
    expect(normalizeDriverType('open-gauss')).toBe('opengauss');
    expect(normalizeDriverType('gauss-db')).toBe('gaussdb');
--- a/frontend/src/utils/connectionDriverType.ts
+++ b/frontend/src/utils/connectionDriverType.ts
@@ -18,6 +18,7 @@ export const normalizeDriverType = (value: string): string => {
  if (normalized === 'chromadb' || normalized === 'chroma-db') return 'chroma';
  if (normalized === 'qdrantdb' || normalized === 'qdrant-db') return 'qdrant';
  if (normalized === 'apache-iotdb' || normalized === 'apache_iotdb') return 'iotdb';
+  if (normalized === 'apache-kafka' || normalized === 'apache_kafka') return 'kafka';
  if (normalized === 'doris') return 'diros';
  if (
    normalized === 'open_gauss' ||
--- a/frontend/src/utils/connectionModalPresentation.test.ts
+++ b/frontend/src/utils/connectionModalPresentation.test.ts
@@ -93,6 +93,7 @@ describe('connectionModalPresentation', () => {
      'redis',
      'tdengine',
      'iotdb',
+      'kafka',
      'custom',
      'jvm',
    ];
@@ -192,6 +193,15 @@ describe('connectionModalPresentation', () => {
      'credentials',
      'databaseScope',
    ]);
+    expect(resolveConnectionConfigLayout('kafka').sections).toEqual([
+      'identity',
+      'uri',
+      'target',
+      'connectionMode',
+      'replica',
+      'service',
+      'credentials',
+    ]);
  });

  it('uses localized labels for layout kinds shown in the modal', () => {
--- a/frontend/src/utils/connectionModalPresentation.ts
+++ b/frontend/src/utils/connectionModalPresentation.ts
@@ -282,6 +282,20 @@ export const resolveConnectionConfigLayout = (
      ],
    };
  }
+  if (type === 'kafka') {
+    return {
+      kind: 'generic-sql',
+      sections: [
+        'identity',
+        'uri',
+        'target',
+        'connectionMode',
+        'replica',
+        'service',
+        'credentials',
+      ],
+    };
+  }
  if (postgresCompatibleTypes.has(type)) {
    return {
      kind: 'postgres-compatible',
--- a/frontend/src/utils/connectionTypeCapabilities.test.ts
+++ b/frontend/src/utils/connectionTypeCapabilities.test.ts
@@ -31,6 +31,7 @@ describe('connectionTypeCapabilities', () => {
    expect(supportsSSLForType('gaussdb')).toBe(true);
    expect(supportsSSLForType('chroma')).toBe(true);
    expect(supportsSSLForType('qdrant')).toBe(true);
+    expect(supportsSSLForType('kafka')).toBe(true);
    expect(supportsSSLForType('tdengine')).toBe(true);
    expect(supportsSSLForType('iotdb')).toBe(false);
    expect(supportsSSLForType('dameng')).toBe(true);
@@ -50,6 +51,8 @@ describe('connectionTypeCapabilities', () => {
    expect(supportsSSLClientCertificateForType('chroma')).toBe(false);
    expect(supportsSSLCAPathForType('qdrant')).toBe(true);
    expect(supportsSSLClientCertificateForType('qdrant')).toBe(false);
+    expect(supportsSSLCAPathForType('kafka')).toBe(true);
+    expect(supportsSSLClientCertificateForType('kafka')).toBe(true);
  });

  it('detects postgres-compatible SSL parameter dialects', () => {
@@ -82,6 +85,7 @@ describe('connectionTypeCapabilities', () => {
    expect(supportsConnectionParamsForType('elasticsearch')).toBe(true);
    expect(supportsConnectionParamsForType('chroma')).toBe(true);
    expect(supportsConnectionParamsForType('qdrant')).toBe(true);
+    expect(supportsConnectionParamsForType('kafka')).toBe(true);
    expect(supportsConnectionParamsForType('redis')).toBe(false);
    expect(supportsConnectionParamsForType('sqlite')).toBe(false);
    expect(supportsConnectionParamsForType('jvm')).toBe(false);
--- a/frontend/src/utils/connectionTypeCapabilities.ts
+++ b/frontend/src/utils/connectionTypeCapabilities.ts
@@ -47,6 +47,7 @@ const sslSupportedTypes = new Set([
  "elasticsearch",
  "chroma",
  "qdrant",
+  "kafka",
 ]);

 export const supportsSSLForType = (type: string) =>
@@ -72,6 +73,7 @@ const sslCAPathSupportedTypes = new Set([
  "elasticsearch",
  "chroma",
  "qdrant",
+  "kafka",
 ]);

 const sslClientCertificateSupportedTypes = new Set([
@@ -91,6 +93,7 @@ const sslClientCertificateSupportedTypes = new Set([
  "gaussdb",
  "mongodb",
  "redis",
+  "kafka",
 ]);

 export const supportsSSLCAPathForType = (type: string) =>
@@ -139,4 +142,5 @@ export const supportsConnectionParamsForType = (type: string) =>
  type === "iotdb" ||
  type === "elasticsearch" ||
  type === "chroma" ||
-  type === "qdrant";
+  type === "qdrant" ||
+  type === "kafka";
--- a/frontend/src/utils/connectionTypeCatalog.test.ts
+++ b/frontend/src/utils/connectionTypeCatalog.test.ts
@@ -15,6 +15,7 @@ describe('connectionTypeCatalog', () => {
      'NoSQL',
      '向量数据库',
      '时序数据库',
+      '消息队列',
      '其他',
    ]);

@@ -28,6 +29,7 @@ describe('connectionTypeCatalog', () => {
    expect(keys).toContain('chroma');
    expect(keys).toContain('qdrant');
    expect(keys).toContain('iotdb');
+    expect(keys).toContain('kafka');
    expect(keys).toContain('jvm');
    expect(keys).toContain('custom');
    expect(new Set(keys).size).toBe(keys.length);
@@ -46,6 +48,7 @@ describe('connectionTypeCatalog', () => {
    expect(getConnectionTypeDefaultPort('chroma')).toBe(8000);
    expect(getConnectionTypeDefaultPort('qdrant')).toBe(6333);
    expect(getConnectionTypeDefaultPort('iotdb')).toBe(6667);
+    expect(getConnectionTypeDefaultPort('kafka')).toBe(9092);
    expect(getConnectionTypeDefaultPort('sqlite')).toBe(0);
    expect(getConnectionTypeDefaultPort('duckdb')).toBe(0);
    expect(getConnectionTypeDefaultPort('unknown')).toBe(3306);
@@ -58,6 +61,7 @@ describe('connectionTypeCatalog', () => {
    expect(getConnectionTypeHint('chroma')).toContain('向量');
    expect(getConnectionTypeHint('qdrant')).toContain('Payload');
    expect(getConnectionTypeHint('iotdb')).toContain('Timeseries');
+    expect(getConnectionTypeHint('kafka')).toContain('Consumer Group');
    expect(getConnectionTypeHint('oceanbase')).toBe('MySQL / Oracle 租户');
    expect(getConnectionTypeHint('duckdb')).toBe('本地文件连接');
    expect(getConnectionTypeHint('mysql')).toBe('标准连接配置');
--- a/frontend/src/utils/connectionTypeCatalog.ts
+++ b/frontend/src/utils/connectionTypeCatalog.ts
@@ -60,6 +60,12 @@ export const CONNECTION_TYPE_GROUPS: ConnectionTypeCatalogGroup[] = [
      { key: 'iotdb', name: 'Apache IoTDB' },
    ],
  },
+  {
+    label: '消息队列',
+    items: [
+      { key: 'kafka', name: 'Kafka' },
+    ],
+  },
  {
    label: '其他',
    items: [
@@ -113,6 +119,8 @@ export const getConnectionTypeDefaultPort = (type: string): number => {
      return 8000;
    case 'qdrant':
      return 6333;
+    case 'kafka':
+      return 9092;
    case 'highgo':
      return 5866;
    case 'mariadb':
@@ -145,6 +153,8 @@ export const getConnectionTypeHint = (type: string): string => {
      return 'Collection 浏览、向量搜索和 Payload 过滤';
    case 'iotdb':
      return 'Storage Group / Device / Timeseries';
+    case 'kafka':
+      return 'Broker / Topic / Consumer Group';
    case 'oceanbase':
      return 'MySQL / Oracle 租户';
    case 'sqlite':
--- a/frontend/src/utils/dataSourceCapabilities.test.ts
+++ b/frontend/src/utils/dataSourceCapabilities.test.ts
@@ -146,6 +146,24 @@ describe('dataSourceCapabilities', () => {
    });
  });

+  it('treats Kafka as a queryable read-only messaging datasource', () => {
+    expect(getDataSourceCapabilities({ type: 'kafka' })).toMatchObject({
+      type: 'kafka',
+      supportsQueryEditor: true,
+      supportsSqlQueryExport: false,
+      supportsCopyInsert: false,
+      supportsCreateDatabase: false,
+      supportsRenameDatabase: false,
+      supportsDropDatabase: false,
+      forceReadOnlyQueryResult: true,
+    });
+    expect(getDataSourceCapabilities({ type: 'custom', driver: 'apache-kafka' })).toMatchObject({
+      type: 'kafka',
+      supportsQueryEditor: true,
+      forceReadOnlyQueryResult: true,
+    });
+  });
+
  it('treats OceanBase Oracle protocol as Oracle capabilities', () => {
    expect(getDataSourceCapabilities({
      type: 'oceanbase',
--- a/frontend/src/utils/dataSourceCapabilities.ts
+++ b/frontend/src/utils/dataSourceCapabilities.ts
@@ -34,6 +34,10 @@ const normalizeDataSourceToken = (raw: string): string => {
    case 'apache-iotdb':
    case 'apache_iotdb':
      return 'iotdb';
+    case 'kafka':
+    case 'apache-kafka':
+    case 'apache_kafka':
+      return 'kafka';
    case 'intersystems':
    case 'intersystemsiris':
    case 'inter-systems':
@@ -107,7 +111,7 @@ const COPY_INSERT_TYPES = new Set([
 ]);

 const QUERY_EDITOR_DISABLED_TYPES = new Set(['redis']);
-const FORCE_READ_ONLY_QUERY_TYPES = new Set(['tdengine', 'iotdb', 'clickhouse']);
+const FORCE_READ_ONLY_QUERY_TYPES = new Set(['tdengine', 'iotdb', 'clickhouse', 'kafka']);
 const MANUAL_TOTAL_COUNT_TYPES = new Set(['duckdb', 'oracle']);
 const APPROXIMATE_TABLE_COUNT_TYPES = new Set(['duckdb', 'oracle']);
 const APPROXIMATE_TOTAL_PAGE_TYPES = new Set(['duckdb']);
--- a/frontend/src/utils/objectQueryTemplates.test.ts
+++ b/frontend/src/utils/objectQueryTemplates.test.ts
@@ -6,4 +6,8 @@ describe('buildTableSelectQuery', () => {
  it('quotes uppercase postgres table names in new query templates', () => {
    expect(buildTableSelectQuery('postgres', 'public.MyTable')).toBe('SELECT * FROM public."MyTable";');
  });
+
+  it('adds a preview limit for Kafka topic browsing', () => {
+    expect(buildTableSelectQuery('kafka', 'logs.app-1')).toBe('SELECT * FROM "logs.app-1" LIMIT 100;');
+  });
 });
--- a/frontend/src/utils/objectQueryTemplates.ts
+++ b/frontend/src/utils/objectQueryTemplates.ts
@@ -5,5 +5,8 @@ export const buildTableSelectQuery = (dbType: string, tableName: string): string
  if (!normalizedTableName) {
    return 'SELECT * FROM ';
  }
+  if (String(dbType || '').trim().toLowerCase() === 'kafka') {
+    return `SELECT * FROM ${quoteQualifiedIdent(dbType, normalizedTableName)} LIMIT 100;`;
+  }
  return `SELECT * FROM ${quoteQualifiedIdent(dbType, normalizedTableName)};`;
 };
--- a/frontend/src/utils/sql.test.ts
+++ b/frontend/src/utils/sql.test.ts
@@ -59,6 +59,11 @@ describe('quoteQualifiedIdent', () => {
      .toBe('`root`.`sg`.`d1`');
  });

+  it('keeps Kafka topic names as one quoted identifier', () => {
+    expect(quoteQualifiedIdent('kafka', 'logs.app-1'))
+      .toBe('"logs.app-1"');
+  });
+
  it('does not split dots inside quoted DuckDB identifiers', () => {
    expect(quoteQualifiedIdent('duckdb', '"daily.events"."2026.06"'))
      .toBe('"daily.events"."2026.06"');
--- a/frontend/src/utils/sql.ts
+++ b/frontend/src/utils/sql.ts
@@ -54,6 +54,9 @@ export const quoteIdentPart = (dbType: string, ident: string) => {
 export const quoteQualifiedIdent = (dbType: string, ident: string) => {
  const raw = (ident || '').trim();
  if (!raw) return raw;
+  if ((dbType || '').trim().toLowerCase() === 'kafka') {
+    return quoteIdentPart(dbType, raw);
+  }
  const parts = splitQualifiedNameSegments(raw).filter(Boolean);
  if (parts.length === 0) return quoteIdentPart(dbType, raw);
  if (parts.length === 1 && parts[0] === normalizeIdentPart(raw)) return quoteIdentPart(dbType, raw);
--- a/frontend/src/utils/sqlDialect.test.ts
+++ b/frontend/src/utils/sqlDialect.test.ts
@@ -38,6 +38,8 @@ describe('sqlDialect', () => {
    expect(resolveSqlDialect('custom', 'qdrant-db')).toBe('qdrant');
    expect(resolveSqlDialect('Apache-IoTDB')).toBe('iotdb');
    expect(resolveSqlDialect('custom', 'apache_iotdb')).toBe('iotdb');
+    expect(resolveSqlDialect('Apache-Kafka')).toBe('kafka');
+    expect(resolveSqlDialect('custom', 'apache_kafka')).toBe('kafka');
    expect(resolveSqlDialect('OceanBase', '', { oceanBaseProtocol: 'oracle' })).toBe('oracle');
    expect(resolveSqlDialect('custom', 'oceanbase', { oceanBaseProtocol: 'oracle' })).toBe('oracle');
    expect(isMysqlFamilyDialect('mariadb')).toBe(true);
@@ -71,6 +73,11 @@ describe('sqlDialect', () => {
    expect(resolveSqlKeywords('iotdb')).not.toEqual(expect.arrayContaining(['TAGS', 'USING']));
  });

+  it('resolves Kafka completion keywords for topic discovery and consume syntax', () => {
+    expect(resolveSqlKeywords('kafka')).toEqual(expect.arrayContaining(['SHOW TOPICS', 'DESCRIBE TOPIC', 'CONSUME']));
+    expect(resolveSqlKeywords('kafka')).not.toEqual(expect.arrayContaining(['ALIGN BY DEVICE', 'AUTO_INCREMENT']));
+  });
+
  it('resolves GaussDB completion keywords and functions as a PostgreSQL-like dialect', () => {
    expect(resolveSqlKeywords('gaussdb')).toEqual(expect.arrayContaining(['RETURNING', 'SERIAL', 'JSONB']));
    expect(names(resolveSqlFunctions('gaussdb'))).toEqual(expect.arrayContaining(['STRING_AGG', 'TO_CHAR', 'CURRENT_DATABASE']));
--- a/frontend/src/utils/sqlDialect.ts
+++ b/frontend/src/utils/sqlDialect.ts
@@ -29,6 +29,7 @@ export type SqlDialect =
  | 'clickhouse'
  | 'tdengine'
  | 'iotdb'
+  | 'kafka'
  | 'mongodb'
  | 'redis'
  | 'elasticsearch'
@@ -135,6 +136,10 @@ export const resolveSqlDialect = (
    case 'apache-iotdb':
    case 'apache_iotdb':
      return 'iotdb';
+    case 'kafka':
+    case 'apache-kafka':
+    case 'apache_kafka':
+      return 'kafka';
    default:
      break;
  }
@@ -159,6 +164,7 @@ export const resolveSqlDialect = (
  if (source.includes('clickhouse')) return 'clickhouse';
  if (source.includes('tdengine')) return 'tdengine';
  if (source.includes('iotdb')) return 'iotdb';
+  if (source.includes('kafka')) return 'kafka';
  if (source.includes('sqlserver') || source.includes('mssql')) return 'sqlserver';
  if (source.includes('iris') || source.includes('intersystems')) return 'iris';
  if (source.includes('elastic')) return 'elasticsearch';
@@ -611,6 +617,17 @@ const IOTDB_KEYWORDS = [
  'COMPRESSION',
 ];

+const KAFKA_KEYWORDS = [
+  'SHOW TOPICS',
+  'SHOW TOPIC',
+  'DESCRIBE TOPIC',
+  'CONSUME',
+  'GROUP',
+  'FROM',
+  'LIMIT',
+  'OFFSET',
+];
+
 export const resolveSqlKeywords = (dbType: string): string[] => {
  const dialect = resolveSqlDialect(dbType);
  if (dialect === 'starrocks') return unique([...COMMON_KEYWORDS, ...MYSQL_KEYWORDS, ...STARROCKS_KEYWORDS]);
@@ -623,6 +640,7 @@ export const resolveSqlKeywords = (dbType: string): string[] => {
  if (dialect === 'clickhouse') return unique([...COMMON_KEYWORDS, ...CLICKHOUSE_KEYWORDS]);
  if (dialect === 'tdengine') return unique([...COMMON_KEYWORDS, ...TDENGINE_KEYWORDS]);
  if (dialect === 'iotdb') return unique([...COMMON_KEYWORDS, ...IOTDB_KEYWORDS]);
+  if (dialect === 'kafka') return unique([...COMMON_KEYWORDS, ...KAFKA_KEYWORDS]);
  return COMMON_KEYWORDS;
 };

--- a/go.mod
+++ b/go.mod
@@ -42,6 +42,7 @@ require (
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/segmentio/encoding v0.5.4 // indirect
+	github.com/segmentio/kafka-go v0.4.51 // indirect
 	github.com/tjfoc/gmsm v1.4.1 // indirect
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
--- a/go.sum
+++ b/go.sum
@@ -258,6 +258,8 @@ github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0=
 github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
 github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0=
 github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0=
+github.com/segmentio/kafka-go v0.4.51 h1:JgDPPG75tC1rWIS2Me6MwcvXJ6f49UQ4HjAOef71Hno=
+github.com/segmentio/kafka-go v0.4.51/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E=
 github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
 github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
 github.com/sijms/go-ora/v2 v2.9.0 h1:+iQbUeTeCOFMb5BsOMgUhV8KWyrv9yjKpcK4x7+MFrg=
--- a/internal/app/db_context.go
+++ b/internal/app/db_context.go
@@ -16,6 +16,8 @@ func normalizeRunConfig(config connection.ConnectionConfig, dbName string) conne
 	}

 	switch strings.ToLower(strings.TrimSpace(config.Type)) {
+	case "kafka", "apache-kafka", "apache_kafka":
+		// Kafka 的 Database 字段表示默认 Topic，不能被树上的 synthetic database(topics) 覆盖。
 	case "oceanbase":
 		if !isOceanBaseOracleProtocol(config) {
 			runConfig.Database = name
@@ -51,7 +53,7 @@ func normalizeSchemaAndTable(config connection.ConnectionConfig, dbName string,

 	// Elasticsearch：索引名可能含多个点（如 iot_pro_biz_operate_log.index.20240626），
 	// 不能按点分割，直接返回原始数据库名和完整表名。
-	if dbType == "elasticsearch" || dbType == "iotdb" {
+	if dbType == "elasticsearch" || dbType == "iotdb" || dbType == "kafka" {
 		return rawDB, rawTable
 	}

@@ -110,6 +112,8 @@ func normalizeSchemaAndTable(config connection.ConnectionConfig, dbName string,
 func normalizeMetadataSchemaAndTable(config connection.ConnectionConfig, dbName string, tableName string) (string, string) {
 	schema, table := normalizeSchemaAndTable(config, dbName, tableName)
 	switch resolveDDLDBType(config) {
+	case "kafka":
+		return schema, table
 	case "postgres", "kingbase", "highgo", "vastbase", "opengauss", "gaussdb":
 		rawTable := strings.TrimSpace(tableName)
 		if rawTable == "" {
--- a/internal/app/db_context_test.go
+++ b/internal/app/db_context_test.go
@@ -242,6 +242,19 @@ func TestNormalizeRunConfig_RedisAllowsDatabaseIndexAboveDefault(t *testing.T) {
 	}
 }

+func TestNormalizeRunConfig_KafkaKeepsDefaultTopic(t *testing.T) {
+	t.Parallel()
+
+	runConfig := normalizeRunConfig(connection.ConnectionConfig{
+		Type:     "kafka",
+		Database: "orders.events",
+	}, "topics")
+
+	if runConfig.Database != "orders.events" {
+		t.Fatalf("expected Kafka default topic to stay orders.events, got %q", runConfig.Database)
+	}
+}
+
 func TestNormalizeSchemaAndTable_IRISDoesNotTreatNamespaceAsSchema(t *testing.T) {
 	t.Parallel()

@@ -294,6 +307,30 @@ func TestNormalizeSchemaAndTable_DuckDBPreservesQuotedQualifiedName(t *testing.T
 	}
 }

+func TestNormalizeSchemaAndTable_KafkaPreservesDottedTopicName(t *testing.T) {
+	t.Parallel()
+
+	schemaOrDb, table := normalizeSchemaAndTable(connection.ConnectionConfig{
+		Type: "kafka",
+	}, "topics", "orders.events.v1")
+
+	if schemaOrDb != "topics" || table != "orders.events.v1" {
+		t.Fatalf("expected kafka topic to stay intact, got %q.%q", schemaOrDb, table)
+	}
+}
+
+func TestNormalizeMetadataSchemaAndTable_KafkaPreservesDottedTopicName(t *testing.T) {
+	t.Parallel()
+
+	schemaOrDb, table := normalizeMetadataSchemaAndTable(connection.ConnectionConfig{
+		Type: "kafka",
+	}, "topics", "logs.app-1")
+
+	if schemaOrDb != "topics" || table != "logs.app-1" {
+		t.Fatalf("expected kafka metadata topic to stay intact, got %q.%q", schemaOrDb, table)
+	}
+}
+
 func TestQuoteTableIdentByType_KingbaseNormalizesQuotedQualifiedTable(t *testing.T) {
 	t.Parallel()

--- a/internal/app/db_proxy.go
+++ b/internal/app/db_proxy.go
@@ -239,6 +239,8 @@ func defaultPortByType(driverType string) int {
 		return 8000
 	case "qdrant":
 		return 6333
+	case "kafka":
+		return 9092
 	default:
 		return 0
 	}
--- a/internal/app/methods_db.go
+++ b/internal/app/methods_db.go
@@ -309,7 +309,7 @@ func normalizeSchemaAndTableByType(dbType string, dbName string, tableName strin
 	}

 	// Elasticsearch：索引名可能含多个点，不能按点分割
-	if dbType == "elasticsearch" {
+	if dbType == "elasticsearch" || dbType == "kafka" {
 		return rawDB, rawTable
 	}

--- a/internal/app/methods_db_create_statement_test.go
+++ b/internal/app/methods_db_create_statement_test.go
@@ -143,6 +143,15 @@ func TestNormalizeSchemaAndTableByType_PGLikeQuotedQualifiedName(t *testing.T) {
 	}
 }

+func TestNormalizeSchemaAndTableByType_KafkaPreservesDottedTopicName(t *testing.T) {
+	t.Parallel()
+
+	schema, table := normalizeSchemaAndTableByType("kafka", "topics", "orders.events.v1")
+	if schema != "topics" || table != "orders.events.v1" {
+		t.Fatalf("expected kafka topic to stay intact, got %q.%q", schema, table)
+	}
+}
+
 func TestBuildRunConfigForDDL_CustomHighGoUsesDatabase(t *testing.T) {
 	t.Parallel()

--- a/internal/app/methods_driver.go
+++ b/internal/app/methods_driver.go
@@ -1427,6 +1427,8 @@ func normalizeDriverType(driverType string) string {
 		return "opengauss"
 	case "gaussdb", "gauss_db", "gauss-db":
 		return "gaussdb"
+	case "kafka", "apache-kafka", "apache_kafka":
+		return "kafka"
 	case "intersystems", "intersystemsiris", "inter-systems-iris", "inter-systems":
 		return "iris"
 	default:
@@ -1495,6 +1497,7 @@ func allDriverDefinitionsWithPackages(packages map[string]pinnedDriverPackage) [
 		{Type: "oracle", Name: "Oracle", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "redis", Name: "Redis", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "postgres", Name: "PostgreSQL", Engine: driverEngineGo, BuiltIn: true},
+		{Type: "kafka", Name: "Kafka", Engine: driverEngineGo, BuiltIn: true},

 		// 其他数据源需要先在驱动管理中“安装启用”。
 		buildOptionalGoDriverDefinition("mariadb", "MariaDB", packages),
--- a/internal/app/methods_driver_version_test.go
+++ b/internal/app/methods_driver_version_test.go
@@ -502,6 +502,22 @@ func TestIoTDBDriverDefinitionUsesOptionalAgent(t *testing.T) {
 	}
 }

+func TestKafkaDriverDefinitionIsBuiltIn(t *testing.T) {
+	definition, ok := resolveDriverDefinition("apache-kafka")
+	if !ok {
+		t.Fatal("expected kafka driver definition")
+	}
+	if definition.Name != "Kafka" {
+		t.Fatalf("unexpected kafka driver name: %q", definition.Name)
+	}
+	if !definition.BuiltIn {
+		t.Fatal("expected kafka to be a built-in driver")
+	}
+	if definition.PinnedVersion != "" || definition.DefaultDownloadURL != "" {
+		t.Fatalf("expected kafka builtin definition to omit optional-agent metadata: %#v", definition)
+	}
+}
+
 func TestGaussDBDriverDefinitionUsesOptionalAgent(t *testing.T) {
 	definition, ok := resolveDriverDefinition("gaussdb")
 	if !ok {
--- a/internal/app/sql_sanitize.go
+++ b/internal/app/sql_sanitize.go
@@ -346,7 +346,7 @@ func isReadOnlySQLQuery(dbType string, query string) bool {
 		return false
 	}
 	switch keyword {
-	case "select", "with", "show", "describe", "desc", "explain", "pragma", "values":
+	case "select", "with", "show", "describe", "desc", "explain", "pragma", "values", "consume":
 		return true
 	default:
 		return false
--- a/internal/app/sql_sanitize_test.go
+++ b/internal/app/sql_sanitize_test.go
@@ -86,6 +86,12 @@ func TestIsReadOnlySQLQuery_ClassifiesWithByTopLevelOperation(t *testing.T) {
 	}
 }

+func TestIsReadOnlySQLQuery_TreatsKafkaConsumeAsReadOnly(t *testing.T) {
+	if !isReadOnlySQLQuery("kafka", `CONSUME GROUP "analytics" FROM "orders.events" LIMIT 20`) {
+		t.Fatal("Kafka CONSUME should be treated as read-only")
+	}
+}
+
 func TestIsBatchableWriteSQLStatement_OnlyMatchesRealWriteStatements(t *testing.T) {
 	if !isBatchableWriteSQLStatement("mysql", "INSERT INTO demo(id) VALUES (1)") {
 		t.Fatal("expected INSERT to be treated as batchable write")
--- a/internal/db/database.go
+++ b/internal/db/database.go
@@ -486,6 +486,9 @@ var databaseFactories = map[string]databaseFactory{
 	"qdrant": func() Database {
 		return &QdrantDB{}
 	},
+	"kafka": func() Database {
+		return &KafkaDB{}
+	},
 }

 func init() {
@@ -524,6 +527,8 @@ func normalizeDatabaseType(dbType string) string {
 		return "chroma"
 	case "qdrantdb", "qdrant-db":
 		return "qdrant"
+	case "kafka", "apache-kafka", "apache_kafka":
+		return "kafka"
 	default:
 		return normalized
 	}
--- a/internal/db/driver_support.go
+++ b/internal/db/driver_support.go
@@ -18,6 +18,7 @@ var coreBuiltinDrivers = map[string]struct{}{
 	"postgres": {},
 	"chroma":   {},
 	"qdrant":   {},
+	"kafka":    {},
 }

 // optionalGoDrivers 表示需要用户“安装启用”后才能使用的纯 Go 驱动。
@@ -78,6 +79,8 @@ func normalizeRuntimeDriverType(driverType string) string {
 		return "qdrant"
 	case "apache-iotdb", "apache_iotdb", "iotdb":
 		return "iotdb"
+	case "kafka", "apache-kafka", "apache_kafka":
+		return "kafka"
 	default:
 		return normalized
 	}
@@ -137,6 +140,8 @@ func driverDisplayName(driverType string) string {
 		return "Chroma"
 	case "qdrant":
 		return "Qdrant"
+	case "kafka":
+		return "Kafka"
 	default:
 		return strings.ToUpper(strings.TrimSpace(driverType))
 	}
--- a/internal/db/driver_support_test.go
+++ b/internal/db/driver_support_test.go
@@ -29,6 +29,11 @@ func TestBuiltinLikeDriversRemainAvailable(t *testing.T) {
 	if !supported {
 		t.Fatalf("redis 应始终可用，reason=%s", reason)
 	}
+
+	supported, reason = DriverRuntimeSupportStatus("kafka")
+	if !supported {
+		t.Fatalf("kafka 应始终可用，reason=%s", reason)
+	}
 }

 func TestOptionalDriverAgentRevisionsGeneratedForOptionalDrivers(t *testing.T) {
--- a/internal/db/kafka_impl.go
+++ b/internal/db/kafka_impl.go
--- a/internal/db/kafka_impl_test.go
+++ b/internal/db/kafka_impl_test.go
@@ -0,0 +1,215 @@
+package db
+
+import (
+	"context"
+	"reflect"
+	"strings"
+	"testing"
+
+	"GoNavi-Wails/internal/connection"
+
+	kafka "github.com/segmentio/kafka-go"
+)
+
+type fakeKafkaRuntime struct {
+	listTopicsResult    []kafkaTopicInfo
+	describeResult      kafkaTopicDescription
+	fetchResult         []kafkaMessageRecord
+	publishAffected     int64
+	lastDescribeTopic   string
+	lastFetchRequest    kafkaFetchRequest
+	lastPublishCommand  kafkaPublishCommand
+}
+
+func (f *fakeKafkaRuntime) Close() error { return nil }
+
+func (f *fakeKafkaRuntime) Ping(ctx context.Context) error { return nil }
+
+func (f *fakeKafkaRuntime) ListTopics(ctx context.Context, includeInternal bool) ([]kafkaTopicInfo, error) {
+	return append([]kafkaTopicInfo(nil), f.listTopicsResult...), nil
+}
+
+func (f *fakeKafkaRuntime) DescribeTopic(ctx context.Context, topic string) (kafkaTopicDescription, error) {
+	f.lastDescribeTopic = topic
+	return f.describeResult, nil
+}
+
+func (f *fakeKafkaRuntime) FetchMessages(ctx context.Context, request kafkaFetchRequest) ([]kafkaMessageRecord, error) {
+	f.lastFetchRequest = request
+	return append([]kafkaMessageRecord(nil), f.fetchResult...), nil
+}
+
+func (f *fakeKafkaRuntime) Publish(ctx context.Context, command kafkaPublishCommand) (int64, error) {
+	f.lastPublishCommand = command
+	return f.publishAffected, nil
+}
+
+func TestNormalizeKafkaConfigParsesURIAndParams(t *testing.T) {
+	config := normalizeKafkaConfig(connection.ConnectionConfig{
+		URI:              "kafka://alice:secret@127.0.0.1:9092,127.0.0.2:9093/orders.events?topology=cluster&tls=true&skip_verify=true",
+		ConnectionParams: "groupId=analytics&mechanism=scram-sha-256",
+	})
+
+	if config.Host != "127.0.0.1" || config.Port != 9092 {
+		t.Fatalf("unexpected primary broker: %#v", config)
+	}
+	if !reflect.DeepEqual(config.Hosts, []string{"127.0.0.2:9093"}) {
+		t.Fatalf("unexpected extra brokers: %#v", config.Hosts)
+	}
+	if config.User != "alice" || config.Password != "secret" {
+		t.Fatalf("unexpected credentials: %#v", config)
+	}
+	if config.Database != "orders.events" || config.Topology != "cluster" {
+		t.Fatalf("unexpected topic/topology: %#v", config)
+	}
+	if !config.UseSSL || config.SSLMode != "skip-verify" {
+		t.Fatalf("unexpected tls settings: %#v", config)
+	}
+
+	params := kafkaConnectionParams(config)
+	if params.Get("groupId") != "analytics" || params.Get("mechanism") != "scram-sha-256" {
+		t.Fatalf("unexpected kafka params: %#v", params)
+	}
+}
+
+func TestKafkaQueryShowTopicsAndDescribeTopic(t *testing.T) {
+	runtime := &fakeKafkaRuntime{
+		listTopicsResult: []kafkaTopicInfo{
+			{Name: "logs.app", Partitions: []kafka.Partition{{}, {}}},
+			{Name: "orders-events", Partitions: []kafka.Partition{{}}},
+		},
+		describeResult: kafkaTopicDescription{
+			Name: "logs.app",
+			Partitions: []kafkaTopicPartition{{
+				ID:               0,
+				Leader:           kafka.Broker{Host: "127.0.0.1", Port: 9092},
+				EarliestOffset:   1,
+				LatestOffset:     9,
+				ApproximateCount: 8,
+			}},
+		},
+	}
+	client := &KafkaDB{runtime: runtime}
+
+	rows, columns, err := client.Query(`SHOW TOPICS LIMIT 1`)
+	if err != nil {
+		t.Fatalf("SHOW TOPICS failed: %v", err)
+	}
+	if len(rows) != 1 || rows[0]["topic"] != "logs.app" {
+		t.Fatalf("unexpected topic rows: %#v", rows)
+	}
+	if !containsString(columns, "partition_count") {
+		t.Fatalf("expected partition_count column, got %v", columns)
+	}
+
+	rows, columns, err = client.Query(`DESCRIBE TOPIC "logs.app"`)
+	if err != nil {
+		t.Fatalf("DESCRIBE TOPIC failed: %v", err)
+	}
+	if runtime.lastDescribeTopic != "logs.app" {
+		t.Fatalf("expected describe topic logs.app, got %q", runtime.lastDescribeTopic)
+	}
+	if len(rows) != 1 || rows[0]["leader"] != "127.0.0.1:9092" {
+		t.Fatalf("unexpected describe rows: %#v", rows)
+	}
+	if !containsString(columns, "approximate_count") {
+		t.Fatalf("expected approximate_count column, got %v", columns)
+	}
+}
+
+func TestKafkaQuerySelectAndConsumeKeepTopicNameIntact(t *testing.T) {
+	runtime := &fakeKafkaRuntime{
+		fetchResult: []kafkaMessageRecord{{
+			Message: kafka.Message{
+				Topic:         "logs.app-1",
+				Partition:     2,
+				Offset:        42,
+				HighWaterMark: 100,
+				Key:           []byte(`{"tenant":"a"}`),
+				Value:         []byte(`{"event":"login","meta":{"ip":"127.0.0.1"}}`),
+			},
+			Key: map[string]interface{}{"tenant": "a"},
+			Value: map[string]interface{}{
+				"event": "login",
+				"meta":  map[string]interface{}{"ip": "127.0.0.1"},
+			},
+			Headers: map[string]interface{}{"x-trace-id": "trace-1"},
+		}},
+	}
+	client := &KafkaDB{
+		runtime:      runtime,
+		defaultGroup: "gonavi",
+		startLatest:  false,
+	}
+
+	rows, columns, err := client.Query(`SELECT * FROM "logs.app-1" LIMIT 5 OFFSET 2`)
+	if err != nil {
+		t.Fatalf("SELECT failed: %v", err)
+	}
+	if runtime.lastFetchRequest.Topic != "logs.app-1" || runtime.lastFetchRequest.Limit != 5 || runtime.lastFetchRequest.Offset != 2 {
+		t.Fatalf("unexpected select fetch request: %#v", runtime.lastFetchRequest)
+	}
+	if len(rows) != 1 || rows[0]["value.meta.ip"] != "127.0.0.1" || rows[0]["headers.x-trace-id"] != "trace-1" {
+		t.Fatalf("unexpected select rows: %#v", rows)
+	}
+	if !containsString(columns, "value.meta.ip") || !containsString(columns, "headers.x-trace-id") {
+		t.Fatalf("unexpected columns: %v", columns)
+	}
+
+	_, _, err = client.Query(`CONSUME FROM "logs.app-1" LIMIT 3`)
+	if err != nil {
+		t.Fatalf("CONSUME failed: %v", err)
+	}
+	if runtime.lastFetchRequest.Topic != "logs.app-1" || runtime.lastFetchRequest.GroupID != "gonavi" || !runtime.lastFetchRequest.Latest {
+		t.Fatalf("unexpected consume request: %#v", runtime.lastFetchRequest)
+	}
+}
+
+func TestKafkaExecPublishesJSONCommand(t *testing.T) {
+	runtime := &fakeKafkaRuntime{publishAffected: 1}
+	client := &KafkaDB{runtime: runtime, defaultTopic: "orders.events"}
+
+	affected, err := client.Exec(`{"key":{"tenant":"a"},"value":{"id":1},"headers":{"x-env":"dev"}}`)
+	if err != nil {
+		t.Fatalf("Exec failed: %v", err)
+	}
+	if affected != 1 {
+		t.Fatalf("unexpected affected rows: %d", affected)
+	}
+	if runtime.lastPublishCommand.Topic != "orders.events" {
+		t.Fatalf("expected default topic publish, got %#v", runtime.lastPublishCommand)
+	}
+	if valueMap, ok := runtime.lastPublishCommand.Value.(map[string]interface{}); !ok || valueMap["id"] == nil {
+		t.Fatalf("unexpected publish value: %#v", runtime.lastPublishCommand.Value)
+	}
+}
+
+func TestKafkaGetColumnsIncludesDerivedFields(t *testing.T) {
+	runtime := &fakeKafkaRuntime{
+		fetchResult: []kafkaMessageRecord{{
+			Message: kafka.Message{Topic: "orders.events"},
+			Value: map[string]interface{}{
+				"meta": map[string]interface{}{
+					"ip": "127.0.0.1",
+				},
+			},
+			Headers: map[string]interface{}{"x-request-id": "req-1"},
+		}},
+	}
+	client := &KafkaDB{runtime: runtime}
+
+	columns, err := client.GetColumns("topics", "orders.events")
+	if err != nil {
+		t.Fatalf("GetColumns failed: %v", err)
+	}
+	names := make([]string, 0, len(columns))
+	for _, col := range columns {
+		names = append(names, col.Name)
+	}
+	joined := strings.Join(names, ",")
+	for _, want := range []string{"topic", "partition", "offset", "value.meta.ip", "headers.x-request-id"} {
+		if !strings.Contains(joined, want) {
+			t.Fatalf("expected derived column %q in %s", want, joined)
+		}
+	}
+}