fix: 对齐技术架构补齐传输 mTLS 与结构化 JSON 日志（NFR 9 章）

架构核对发现 2 处差距，本次补齐：
- Kafka 上行支持 SSL/SASL_SSL 双向 mTLS（8.2 服务间 mTLS 边缘网关↔总线）
- 网关日志支持结构化 JSON 输出（NFR 可维护：统一日志规范）
- 配置示例与 README 验收口径同步更新

core/data-bus/__init__.py

@@ -0,0 +1,9 @@
+# -*- coding: utf-8 -*-
+"""iAOP-Core · 数据总线 + 时序库（Data Bus）—— 模板化封装。
+
+对应 PRD 5.2「② 数据总线 + 时序库」与 Issue #4（EPIC）：
+复用 Kafka + TDengine + PostgreSQL + MinIO，全部改为**按模板配置**
+topic / 时序库表 / 关系表 / 对象桶，换行业只改模板资产（YAML + 点位字典），
+内核代码零改动；批量写入缓冲 + 幂等去重，保障「数据不丢不重」。
+"""
+__version__ = "0.1.0"

core/data-bus/postgres_schema.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""PostgreSQL 关系表 schema（模板 / 模型 / 用户 / 权限）—— 子任务 #30。
+
+按模板自动生成隔离 schema（`tpl_{tpl}`），内含四张基础关系表：
+- templates   模板注册表（行业模板配置资产）；
+- models      模型注册表（模型版本 / 算法 / 超参包）；
+- users       用户表（角色 + 细粒度权限 JSONB）；
+- permissions 角色-资源-动作权限矩阵。
+
+另生成角色授权语句（GRANT），按模板配置声明 databus_ro / databus_rw 等角色。
+不依赖 psycopg2：仅产出 DDL 文本，供模板配置台预览 / 运维执行。
+"""
+from __future__ import annotations
+
+from typing import List, Optional
+
+from .templating import TemplateNaming
+
+# 默认基础表清单（对应子任务 #30「模板/模型/用户/权限」）
+DEFAULT_TABLES: List[str] = ["templates", "models", "users", "permissions"]
+
+# 表 → 列定义（可被模板 YAML 的 schema_tables 覆盖为子集/扩展）
+TABLE_COLUMNS: dict = {
+    "templates": [
+        "id BIGSERIAL PRIMARY KEY",
+        "template VARCHAR(64) NOT NULL UNIQUE",
+        "version VARCHAR(32) NOT NULL DEFAULT '0.1.0'",
+        "config JSONB NOT NULL DEFAULT '{}'::jsonb",
+        "status VARCHAR(16) NOT NULL DEFAULT 'draft'",
+        "created_at TIMESTAMPTZ NOT NULL DEFAULT now()",
+    ],
+    "models": [
+        "id BIGSERIAL PRIMARY KEY",
+        "model_id VARCHAR(128) NOT NULL UNIQUE",
+        "template VARCHAR(64) REFERENCES {schema}.templates(template) ON DELETE CASCADE",
+        "algorithm VARCHAR(32) NOT NULL DEFAULT 'xgboost'",
+        "hyperparams JSONB NOT NULL DEFAULT '{}'::jsonb",
+        "version VARCHAR(32) NOT NULL DEFAULT '0.1.0'",
+        "status VARCHAR(16) NOT NULL DEFAULT 'staging'",
+        "created_at TIMESTAMPTZ NOT NULL DEFAULT now()",
+    ],
+    "users": [
+        "id BIGSERIAL PRIMARY KEY",
+        "username VARCHAR(64) NOT NULL UNIQUE",
+        "role VARCHAR(32) NOT NULL DEFAULT 'viewer'",
+        "permissions JSONB NOT NULL DEFAULT '[]'::jsonb",
+        "created_at TIMESTAMPTZ NOT NULL DEFAULT now()",
+    ],
+    "permissions": [
+        "id BIGSERIAL PRIMARY KEY",
+        "role VARCHAR(32) NOT NULL",
+        "resource VARCHAR(64) NOT NULL",
+        "action VARCHAR(16) NOT NULL",
+        "UNIQUE (role, resource, action)",
+    ],
+}
+
+
+def generate_schema_ddl(
+    naming: TemplateNaming,
+    tables: Optional[List[str]] = None,
+    table_columns: Optional[dict] = None,
+) -> str:
+    """生成 schema 与基础关系表的完整 DDL（幂等 IF NOT EXISTS）。
+
+    Args:
+        naming: 模板命名器（决定 schema 名）。
+        tables: 表清单（默认 DEFAULT_TABLES）。
+        table_columns: 表→列定义覆盖（默认 TABLE_COLUMNS）。
+
+    Returns:
+        多语句 DDL 文本（含 CREATE SCHEMA / CREATE TABLE）。
+    """
+    names = tables or DEFAULT_TABLES
+    cols = table_columns or TABLE_COLUMNS
+    schema = naming.pg_schema()
+
+    lines = [f"CREATE SCHEMA IF NOT EXISTS {schema};", ""]
+    for table in names:
+        columns = cols.get(table)
+        if not columns:
+            continue
+        lines.append(f"CREATE TABLE IF NOT EXISTS {schema}.{table} (")
+        lines.append("  " + ",\n  ".join(c.format(schema=schema) for c in columns))
+        lines.append(");")
+        lines.append("")
+    return "\n".join(lines).rstrip() + "\n"
+
+
+def generate_grant_ddl(
+    naming: TemplateNaming,
+    roles: Optional[List[str]] = None,
+) -> List[str]:
+    """按模板声明的关系角色生成授权语句。
+
+    角色约定：`{role}_ro` 只读、`{role}_rw` 读写（可扩展）。
+    Args:
+        naming: 模板命名器。
+        roles: 角色清单（如 ["databus"] → databus_ro / databus_rw）。
+    Returns:
+        GRANT 语句列表。
+    """
+    schema = naming.pg_schema()
+    grants: List[str] = []
+    for role in (roles or ["databus"]):
+        for suffix, privileges in (("ro", "SELECT"), ("rw", "SELECT, INSERT, UPDATE, DELETE")):
+            role_name = f"{role}_{suffix}"
+            grants.append(f"GRANT USAGE ON SCHEMA {schema} TO {role_name};")
+            grants.append(
+                f"GRANT {privileges} ON ALL TABLES IN SCHEMA {schema} TO {role_name};"
+            )
+            if suffix == "rw":
+                grants.append(
+                    f"GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA {schema} TO {role_name};"
+                )
+    return grants

core/data-bus/tdengine_schema.py

@@ -0,0 +1,156 @@
+# -*- coding: utf-8 -*-
+"""TDengine 超级表 schema 自动生成 + 批量 INSERT —— 依据点位字典（子任务 #29）。
+
+模型（TDengine 2.0/3.0 通用）：
+- 一张超级表 `{tpl}_points`：列 `ts / value / quality`，标签
+  `device_id / unit / data_type`；
+- 点位字典每个测点自动生成一张子表 `{tpl}_pt_{point_id}`（点位维度，
+  字典 CSV 变更即重建子表集，内核零改动）；
+- 批量写入：按子表聚合多行 `INSERT INTO {sub} VALUES (...),(...);`，
+  单条样本约 512B 时对齐 PRD 5.2「5k 条/100ms」基线。
+
+不依赖 taospy：本模块只负责生成 DDL / 批量 SQL；真实落库由
+`batch_writer.StoreSink` 的客户端实现（缺失时降级内存 sink 联调）。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+from .templating import TemplateNaming, sanitize_sql, _sql_str
+
+# dataType → TDengine 列类型（点位字典 schema.VALID_DATA_TYPES 子集）
+TD_VALUE_TYPE: Dict[str, str] = {
+    "float": "DOUBLE",
+    "int": "BIGINT",
+    "bool": "BOOL",
+}
+DEFAULT_VALUE_TYPE = "DOUBLE"
+
+# 样本记录字段（与 edge-gateway spool / kafka_sink 保持一致）
+ROW_KEYS = ("device_id", "point_id", "value", "ts", "quality")
+
+
+@dataclass(frozen=True)
+class PointSpec:
+    """点位字典中的最小维度信息（schema 生成所需）。"""
+
+    device_id: str
+    point_id: str
+    unit: str = ""
+    data_type: str = "float"
+
+    @classmethod
+    def from_dict(cls, raw: dict) -> "PointSpec":
+        return cls(
+            device_id=str(raw.get("device_id") or ""),
+            point_id=str(raw.get("point_id") or ""),
+            unit=str(raw.get("unit") or ""),
+            data_type=str(raw.get("data_type") or raw.get("dataType") or "float"),
+        )
+
+
+def value_type(data_type: str) -> str:
+    return TD_VALUE_TYPE.get((data_type or "").lower(), DEFAULT_VALUE_TYPE)
+
+
+def generate_supertable_ddl(
+    naming: TemplateNaming,
+    retention_days: Optional[int] = None,
+) -> str:
+    """生成超级表 DDL（自动建表，幂等 IF NOT EXISTS）。
+
+    Args:
+        naming: 模板命名器（决定超级表名）。
+        retention_days: 数据保留天数（TDengine KEEP 表选项，可选）。
+
+    Returns:
+        CREATE STABLE 语句。
+    """
+    ddl = (
+        f"CREATE STABLE IF NOT EXISTS {naming.stable()} (\n"
+        "  ts TIMESTAMP,\n"
+        "  value DOUBLE,\n"
+        "  quality TINYINT\n"
+        ") TAGS (\n"
+        "  device_id NCHAR(64),\n"
+        "  unit NCHAR(16),\n"
+        "  data_type NCHAR(16)\n"
+        ")"
+    )
+    if retention_days and int(retention_days) > 0:
+        ddl += f" KEEP({int(retention_days)})"
+    return ddl + ";"
+
+
+def generate_subtable_ddls(
+    naming: TemplateNaming,
+    points: List[PointSpec],
+    use_typed_value: bool = True,
+) -> List[str]:
+    """按点位字典为每个测点生成子表 DDL（依据点位字典自动生成）。
+
+    Args:
+        naming: 模板命名器。
+        points: 点位字典（测点维度集合）。
+        use_typed_value: 是否按 dataType 派生 value 列类型（TDengine 3.x
+            支持列级类型；2.x 超级表统一 DOUBLE，传 False 时忽略）。
+
+    Returns:
+        CREATE TABLE ... USING ... TAGS(...) 语句列表（顺序与 points 一致）。
+    """
+    if use_typed_value:
+        # 说明：超级表 value 列按最宽类型 DOUBLE 建（TDengine 列类型在
+        # CREATE STABLE 时固定），子表继承，这里保留 data_type 标签供聚合。
+        pass
+    out: List[str] = []
+    seen: set = set()
+    for p in points:
+        sub = naming.subtable(p.point_id)
+        if sub in seen:
+            continue
+        seen.add(sub)
+        tags = ", ".join(
+            _sql_str(v) for v in (p.device_id, p.unit or "", (p.data_type or "float").lower())
+        )
+        out.append(
+            f"CREATE TABLE IF NOT EXISTS {sub} USING {naming.stable()} "
+            f"TAGS ({tags});"
+        )
+    return out
+
+
+def build_batch_insert(
+    naming: TemplateNaming,
+    rows: List[dict],
+) -> List[str]:
+    """把样本批聚合为按子表分组的批量 INSERT 语句。
+
+    Args:
+        naming: 模板命名器（决定子表名）。
+        rows: 样本列表，每条含 device_id/point_id/value/ts(/quality)。
+
+    Returns:
+        SQL 语句列表：每子表一条 `INSERT INTO {sub} VALUES (...),(...);`。
+        空输入返回空列表。
+    """
+    grouped: Dict[str, List[dict]] = {}
+    for row in rows:
+        point_id = str(row.get("point_id") or "")
+        sub = naming.subtable(point_id)
+        grouped.setdefault(sub, []).append(row)
+
+    statements: List[str] = []
+    for sub in sorted(grouped):
+        tuples = []
+        for row in grouped[sub]:
+            ts = row.get("ts")
+            ts_ms = int(ts * 1000) if isinstance(ts, float) else int(ts)
+            value = row.get("value")
+            value_sql = "NULL" if value is None else repr(float(value))
+            quality = int(row.get("quality", 1))
+            tuples.append(f"({ts_ms}, {value_sql}, {quality})")
+        statements.append(
+            f"INSERT INTO {sub} VALUES " + ", ".join(tuples) + ";"
+        )
+    return statements

core/data-bus/templating.py

@@ -0,0 +1,123 @@
+# -*- coding: utf-8 -*-
+"""数据总线命名与分区模板化 —— 换行业只改配置，内核零改动（PRD 5.2 / Issue #4）。
+
+模板（template）是行业复制的唯一粒度，所有外部资源命名均由
+「模板名 + 点位维度」推导：
+
+- Kafka topic ：`{topic_prefix}.{device_id}.points`（与 edge-gateway 上行一致）；
+  分区 = hash(device_id) % num_partitions，保证单设备分区内有序；
+- TDengine    ：超级表 `{tpl}_points`、每测点子表 `{tpl}_pt_{point_id}`；
+- PostgreSQL  ：schema `tpl_{tpl}`，表名 `{schema}.{table}`；
+- MinIO       ：桶 `{template}-artifacts`，对象键 `features/{model_id}/{...}`。
+
+命名清洗规则：
+- Kafka topic / MinIO 桶：小写 + 保留 `-` / `_` / `.` 之外替换为 `_`；
+- SQL 标识符（TDengine / PostgreSQL）：额外把 `-` 替换为 `_`，避免引号转义。
+"""
+from __future__ import annotations
+
+import hashlib
+import re
+from typing import List, Optional
+
+# 保留字符集（Kafka topic / MinIO 桶名均允许小写字母、数字、- _ .）
+_KEEP = re.compile(r"[^a-z0-9_.-]+")
+# SQL 标识符额外排除 `-`（TDengine / PostgreSQL 不加引号时不允许）
+_SQL_KEEP = re.compile(r"[^a-z0-9_.]+")
+
+
+def sanitize(name: str) -> str:
+    """通用命名清洗：小写 + 非安全字符替换为 `_`。"""
+    s = (name or "").strip().lower()
+    s = _KEEP.sub("_", s)
+    s = s.strip("._") or "tpl"
+    return s
+
+
+def sanitize_sql(name: str) -> str:
+    """SQL 标识符清洗：`-` 与其余非安全字符替换为 `_`。"""
+    s = sanitize(name)
+    s = _SQL_KEEP.sub("_", s)
+    return s.strip("._") or "tpl"
+
+
+def _sql_str(value: str) -> str:
+    """SQL 字符串字面量转义（单引号加倍）。"""
+    return "'" + str(value).replace("'", "''") + "'"
+
+
+class TemplateNaming:
+    """按模板推导全部外部资源命名（Kafka / TDengine / PostgreSQL / MinIO）。"""
+
+    def __init__(
+        self,
+        template: str,
+        topic_prefix: Optional[str] = None,
+        num_partitions: int = 12,
+        bucket_suffix: str = "artifacts",
+    ):
+        self.template = sanitize(template)
+        # SQL 标识符用下划线形态（避免 `-` 需加引号）
+        self.tpl_sql = sanitize_sql(self.template)
+        self.topic_prefix = sanitize(topic_prefix) if topic_prefix else self.template
+        self.num_partitions = max(1, int(num_partitions))
+        self.bucket_suffix = sanitize_sql(bucket_suffix)
+
+    # ------------------------------------------------------------------
+    # Kafka：topic 命名 / 分区策略（子任务 #28）
+    # ------------------------------------------------------------------
+    def topic(self, device_id: str) -> str:
+        """上行 topic：`{topic_prefix}.{device_id}.points`（对齐 edge-gateway）。"""
+        return f"{self.topic_prefix}.{sanitize(device_id)}.points"
+
+    def partition(self, device_id: str, num_partitions: Optional[int] = None) -> int:
+        """分区策略：按 device_id 一致性哈希 → 单设备分区内严格有序。"""
+        n = num_partitions or self.num_partitions
+        digest = hashlib.md5(sanitize(device_id).encode("utf-8")).hexdigest()
+        return int(digest[:8], 16) % n
+
+    def partitions(self, device_ids: List[str], num_partitions: Optional[int] = None) -> dict:
+        """设备 → 分区映射（模板配置台预览用）。"""
+        n = num_partitions or self.num_partitions
+        return {d: self.partition(d, n) for d in device_ids}
+
+    # ------------------------------------------------------------------
+    # TDengine：超级表 + 每测点子表（子任务 #29）
+    # ------------------------------------------------------------------
+    def stable(self) -> str:
+        """时序超级表：`{tpl}_points`。"""
+        return f"{self.tpl_sql}_points"
+
+    def subtable(self, point_id: str) -> str:
+        """测点子表：`{tpl}_pt_{point_id}`（点位维度，字典驱动自动生成）。"""
+        return f"{self.tpl_sql}_pt_{sanitize_sql(point_id)}"
+
+    # ------------------------------------------------------------------
+    # PostgreSQL：schema 与表（子任务 #30）
+    # ------------------------------------------------------------------
+    def pg_schema(self) -> str:
+        """关系 schema：`tpl_{tpl}`。"""
+        return f"tpl_{self.tpl_sql}"
+
+    def pg_table(self, table: str) -> str:
+        """`{schema}.{table}` 限定名。"""
+        return f"{self.pg_schema()}.{sanitize_sql(table)}"
+
+    # ------------------------------------------------------------------
+    # MinIO：对象桶与对象键（子任务 #31）
+    # ------------------------------------------------------------------
+    def bucket(self) -> str:
+        """对象存储桶：`{template}-artifacts`（S3 桶名允许 `-`）。"""
+        return f"{self.template}-{self.bucket_suffix}"
+
+    def snapshot_key(self, model_id: str, date: str, seq: int) -> str:
+        """特征快照对象键：`features/{model_id}/{date}/{seq:06d}.jsonl`。"""
+        return f"features/{sanitize_sql(model_id)}/{date}/{int(seq):06d}.jsonl"
+
+    def model_artifact_key(self, model_id: str, version: str) -> str:
+        """模型 artifact 对象键：`models/{model_id}/{version}/model.bin`。"""
+        return f"models/{sanitize_sql(model_id)}/{sanitize_sql(version)}/model.bin"
+
+    # ------------------------------------------------------------------
+    def _sql_str(self, value: str) -> str:
+        return _sql_str(value)

core/edge-gateway/README.md

@@ -12,7 +12,8 @@
 | 采集 P99 延迟 | ≤ 1.8s（600 点位 1Hz） | `collector/metrics.py` 统计，`engine.py` 调度 |
 | 丢失率 | ≤ 0.02% | `collector/spool.py` 断点续传 + 背压丢弃计数 |
 | 可用性 | ≥ 99.8% | `collector/metrics.py` 轮次成功率 |
-| 安全 | 零控制指令下发 | 驱动抽象仅暴露只读读接口（`drivers/base.py`） |
+| 安全-网络 | 严格只读、零控制指令；传输 mTLS | 驱动仅只读接口；`upstream/kafka_sink.py` 支持 SSL/SASL_SSL 双向 mTLS |
+| 可维护 | 统一日志规范（结构化 JSON） | `main.py --log-json` 输出结构化 JSON 日志 |
 
 ## 目录结构
 

core/edge-gateway/config/gateway.example.yaml

@@ -51,3 +51,14 @@ kafka:
   bootstrap_servers: "10.20.0.10:9092"
   topic_prefix: "iaop.ti-cl4"           # 模板化 topic：{prefix}.{device_id}.points
   batch_size: 500
+  # 传输安全（NFR 9 章：服务间 mTLS，边缘网关↔总线双向认证）。
+  # 现场部署必须开启；本地联调可保持 PLAINTEXT。
+  security:
+    protocol: SSL                        # PLAINTEXT | SSL | SASL_SSL
+    ca_location: "/etc/iaop/certs/ca.crt"
+    cert_location: "/etc/iaop/certs/gateway.crt"
+    key_location: "/etc/iaop/certs/gateway.key"
+    # protocol: SASL_SSL 时追加：
+    # sasl_mechanism: "PLAIN"
+    # sasl_username: "edge-gateway"
+    # sasl_password: "..."

core/edge-gateway/main.py

@@ -35,6 +35,8 @@ def parse_args(argv: List[str]) -> argparse.Namespace:
     parser.add_argument("--rounds", type=int, default=0,
                         help="采集轮数上限（0=无限，调试用）")
     parser.add_argument("--verbose", action="store_true", help="输出调试日志")
+    parser.add_argument("--log-json", action="store_true",
+                        help="结构化 JSON 日志（NFR 9 章统一日志规范；生产模式建议开启）")
     return parser.parse_args(argv)
 
 
@@ -78,7 +80,7 @@ def build_engine(config: dict, point_dict: PointDict):
 
 
 def build_sink(config: dict, spool) -> object:
-    """按模板配置组装 Kafka 上行通道。"""
+    """按模板配置组装 Kafka 上行通道（含 mTLS 传输安全，NFR 9 章）。"""
     from upstream import KafkaSink
 
     kafka_cfg = config.get("kafka", {})
@@ -87,11 +89,33 @@ def build_sink(config: dict, spool) -> object:
         topic_prefix=kafka_cfg.get("topic_prefix", "iaop"),
         spool=spool,
         batch_size=int(kafka_cfg.get("batch_size", 500)),
+        security=kafka_cfg.get("security") or {},
     )
 
 
 def main(argv: List[str]) -> int:
     args = parse_args(argv)
+
+    if args.log_json:
+        # NFR 9 章「可维护」：统一日志规范 —— 结构化 JSON，便于采集到统一日志平台
+        import json as _json
+
+        class JsonFormatter(logging.Formatter):
+            def format(self, record):
+                payload = {
+                    "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
+                    "level": record.levelname,
+                    "logger": record.name,
+                    "msg": record.getMessage(),
+                }
+                if record.exc_info:
+                    payload["exc"] = self.formatException(record.exc_info)
+                return _json.dumps(payload, ensure_ascii=False)
+
+        handler = logging.StreamHandler()
+        handler.setFormatter(JsonFormatter())
+        logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, handlers=[handler])
+    else:
         logging.basicConfig(
             level=logging.DEBUG if args.verbose else logging.INFO,
             format="%(asctime)s %(levelname)s %(name)s: %(message)s",

core/edge-gateway/upstream/kafka_sink.py

@@ -29,11 +29,24 @@ class KafkaSink:
         topic_prefix: str,
         spool: SpoolStore,
         batch_size: int = 500,
+        security: Optional[dict] = None,
     ):
+        """
+        Args:
+            bootstrap_servers: Kafka 地址列表（逗号分隔）；
+            topic_prefix: 上行 topic 前缀（模板化 `{prefix}.{device}.points`）；
+            spool: 本地缓存（断点续传）；
+            batch_size: 单批上限；
+            security: 传输安全配置（NFR 9 章：服务间 mTLS）。
+                约定字段：protocol(PLAINTEXT|SSL|SASL_SSL)、ca_location、
+                cert_location、key_location、sasl_username、sasl_password。
+                其中 SSL/SASL_SSL + ca/cert/key = 边缘网关↔总线双向 mTLS。
+        """
         self.bootstrap_servers = bootstrap_servers
         self.topic_prefix = topic_prefix
         self.spool = spool
         self.batch_size = max(1, batch_size)
+        self.security = security or {}
         self._producer = None
         self._degraded = False  # True = Kafka 不可用，仅保留 spool
         self._lock = threading.Lock()
@@ -42,6 +55,27 @@ class KafkaSink:
         self._connect()
 
     # ------------------------------------------------------------------
+    def _producer_config(self) -> dict:
+        """组装 confluent-kafka producer 配置（含 mTLS/SASL 透传）。"""
+        conf = {"bootstrap.servers": self.bootstrap_servers}
+        protocol = self.security.get("protocol", "PLAINTEXT").upper()
+        conf["security.protocol"] = protocol
+        if protocol in ("SSL", "SASL_SSL"):
+            for key, kafka_key in (
+                ("ca_location", "ssl.ca.location"),
+                ("cert_location", "ssl.certificate.location"),
+                ("key_location", "ssl.key.location"),
+                ("key_password", "ssl.key.password"),
+            ):
+                if self.security.get(key):
+                    conf[kafka_key] = self.security[key]
+        if protocol == "SASL_SSL":
+            conf["sasl.mechanism"] = self.security.get("sasl_mechanism", "PLAIN")
+            if self.security.get("sasl_username"):
+                conf["sasl.username"] = self.security["sasl_username"]
+                conf["sasl.password"] = self.security.get("sasl_password", "")
+        return conf
+
     def _connect(self) -> None:
         """尝试连接 Kafka；失败则降级（不阻断采集）。"""
         try:
@@ -51,7 +85,7 @@ class KafkaSink:
             self._degraded = True
             return
         try:
-            self._producer = Producer({"bootstrap.servers": self.bootstrap_servers})
+            self._producer = Producer(self._producer_config())
         except Exception as exc:
             logger.warning("Kafka 初始化失败(%s)，降级为 spool-only 模式", exc)
             self._degraded = True