Apache Airflow配置Kafka连接报错：config['bootstrap.servers']必须提供

以下是针对该问题的具体排查步骤：

• 确认连接类型选择正确
  在Airflow UI创建连接时，必须选择Apache Kafka专属连接类型，而非自定义或其他类型。不同连接类型的参数解析逻辑不同，选错类型会导致bootstrap.servers无法被识别。

• 验证参数输入格式
  bootstrap.servers必须在连接的Extra字段中以标准JSON格式输入，不能填在Host或其他字段里。正确输入示例：

  {"bootstrap.servers": "kafka-broker1:9092,kafka-broker2:9092"}
  

  注意检查JSON语法：引号配对、逗号分隔是否正确，语法错误会直接导致参数解析失败。

• 检查版本兼容性
  确认apache-airflow-providers-apache-kafka版本与Airflow核心版本兼容。可通过以下命令查看当前版本：

  pip show apache-airflow apache-airflow-providers-apache-kafka
  

  若版本不兼容，可升级/降级提供者包：

  pip install apache-airflow-providers-apache-kafka==<兼容版本号>
  

  参考Airflow官方文档的提供者兼容性矩阵确认适配版本。

• 测试连接参数读取情况
  编写简单测试DAG，验证连接配置是否被正确加载：

  from airflow import DAG
  from airflow.providers.apache.kafka.hooks.kafka import KafkaHook
  from airflow.utils.dates import days_ago
  
  with DAG(
      'test_kafka_connection',
      schedule_interval=None,
      start_date=days_ago(1),
  ) as dag:
      def test_kafka_conn():
          hook = KafkaHook(kafka_conn_id='你的Kafka连接ID')
          config = hook.get_conn_config()
          print(f"加载的配置: {config}")
          print(f"bootstrap.servers值: {config.get('bootstrap.servers')}")
  
      test_kafka_conn()
  

  运行DAG查看日志，若未输出bootstrap.servers的值，说明参数未被正确读取。

• 排查全局配置覆盖
  检查Airflow的airflow.cfg文件，确认是否存在全局Kafka配置（如kafka.bootstrap.servers），这类全局配置可能会覆盖连接中的自定义参数。

• 清理缓存并重启服务
  Airflow元数据缓存可能导致参数不生效，尝试重启Webserver和Scheduler：

  airflow webserver restart
  airflow scheduler restart
  

  同时可清理$AIRFLOW_HOME/tmp下的临时缓存文件。

内容的提问来源于stack exchange，提问作者Nikita