Ansible基于set_fact循环重启Kafka Connector失败任务配置方法

核心逻辑是将采集到的「连接器-多失败任务ID」嵌套结构展平为单任务维度的列表，再循环调用重启接口。

具体实现 #

1. 修正采集逻辑的状态筛选问题 #

原有采集任务的JMESPath条件写的是筛选RUNNING状态任务，和重启失败任务的目标不符，先修正set_fact的筛选规则：

- name: 采集失败任务映射关系
      set_fact:
        failed_connector_name_task_id: "{{ conn_stat_1 | json_query('results[].json[].{name: name ,id: tasks[?state == `FAILED`].id}') }}"

2. 补全重启任务循环 #

不需要额外新增中间变量，直接用subelements过滤器在循环时展平嵌套结构即可，完整重启任务代码如下：

- name: 重启所有失败的Connector任务
      uri:
        url: "{{ scheme }}://{{ server }}:{{ port_no }}/connectors/{{ item.0.name }}/tasks/{{ item.1 }}/restart"
        user: "{{ username }}"
        password: "{{ password }}"
        method: POST
        force_basic_auth: yes
        status_code: [200, 202]
      loop: "{{ failed_connector_name_task_id | subelements('id') }}"
      register: restart_op_result
      ignore_errors: yes

逻辑说明 #

• subelements('id')过滤器会遍历每个连接器下的id数组，把每个任务ID和所属连接器信息组合为单独的循环项：
  • item.0对应当前循环项所属的连接器对象，item.0.name取连接器名称
  • item.1对应当前循环项的单个任务ID
• 接口状态码兼容200、202：部分版本的Kafka Connect重启接口为异步执行，会返回202状态码，可根据自身集群版本调整
• 加ignore_errors: yes是为了避免单个任务重启失败中断整个playbook，保证其余失败任务能正常执行重启操作

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