Ansible中Kafka Connect连接器部署任务循环重试及故障主机不跳过的实现方案咨询

看了你的问题，核心痛点是两个：一是检查连接器状态时，因为Kafka Connect返回的JSON里tasks列表为空，导致Ansible条件判断报错中断重试；二是希望即使某个主机在任务中报错，后续任务也能继续在该主机上执行，而不是被跳过。我来给你梳理下可行的解决思路：

问题根源分析 #

你遇到的报错是因为当连接器刚创建时，Kafka Connect还没初始化好任务，返回的tasks是个空列表，这时候直接访问result.json.tasks.0.state就会触发索引越界错误，Ansible会把这个错误视为任务失败，进而终止该主机的后续任务执行，同时也无法继续重试这个任务。

解决方案步骤 #

1. 修复状态检查的条件判断，避免报错并允许正常重试

首先要让条件判断变得“安全”，即使tasks为空或者不存在，也不会抛出错误，而是返回false让until继续重试。你可以用两种方式实现：

方式一：显式判断存在性和长度

修改Check that the connector is up任务的until条件，先确认tasks存在且非空，再判断状态：

- name: Check that the connector is up
  uri:
    url: "http://{{groups['connect_cluster'][0]}}:8083/connectors/{{item.name}}/status"
    method: GET
    return_content: yes
    status_code: 200
    body_format: json
  register: result
  until: >
    result.json is defined and
    'tasks' in result.json and
    result.json.tasks | length > 0 and
    result.json.tasks.0.state == "RUNNING"
  retries: 10
  delay: 1
  changed_when: false

方式二：用Jinja2过滤器做默认值处理

更简洁的写法是用default过滤器处理空值情况，避免索引越界：

- name: Check that the connector is up
  uri:
    url: "http://{{groups['connect_cluster'][0]}}:8083/connectors/{{item.name}}/status"
    method: GET
    return_content: yes
    status_code: 200
    body_format: json
  register: result
  until: >
    (result.json.tasks | default([]) | first | default({})).state == "RUNNING"
  retries: 10
  delay: 1
  changed_when: false

解释：

• result.json.tasks | default([])：如果tasks字段不存在，返回空列表
• | first：取列表第一个元素，空列表会返回None
• | default({})：如果是None，返回空字典
• 最后访问.state，空字典的.state是undefined，和"RUNNING"不相等，条件返回false，触发重试

2. 确保故障主机后续任务不被跳过

现在条件判断不会报错了，但如果重试10次后连接器还是没进入RUNNING状态，任务会标记为失败，Ansible默认会跳过该主机的后续任务。要解决这个问题，你可以添加ignore_errors: true，让任务即使失败也继续执行后续步骤：

- name: Check that the connector is up
  uri:
    url: "http://{{groups['connect_cluster'][0]}}:8083/connectors/{{item.name}}/status"
    method: GET
    return_content: yes
    status_code: 200
    body_format: json
  register: result
  until: >
    (result.json.tasks | default([]) | first | default({})).state == "RUNNING"
  retries: 10
  delay: 1
  ignore_errors: true
  changed_when: false

这样，即使某个主机的连接器状态检查最终失败，后续的连接器部署任务依然会在该主机上执行，不会被跳过。

3. 可选：自定义失败判断逻辑

如果你希望更精细地控制什么时候才算“真正失败”（比如只有当URI请求本身失败，或者重试完状态仍不是RUNNING才标记失败），可以用failed_when替代ignore_errors：

- name: Check that the connector is up
  uri:
    url: "http://{{groups['connect_cluster'][0]}}:8083/connectors/{{item.name}}/status"
    method: GET
    return_content: yes
    status_code: 200
    body_format: json
  register: result
  until: >
    (result.json.tasks | default([]) | first | default({})).state == "RUNNING"
  retries: 10
  delay: 1
  failed_when: >
    result is failed or
    ((result.json.tasks | default([]) | first | default({})).state != "RUNNING")
  changed_when: false
  ignore_errors: true

这里failed_when定义了两种失败场景：一是URI请求本身失败（比如返回非200状态码），二是重试完后状态仍不是RUNNING。结合ignore_errors: true，既能标记失败状态，又不影响后续任务执行。

额外建议 #

你原来的URI请求是固定访问groups['connect_cluster'][0]节点，也就是所有主机都检查同一个节点的连接器状态。如果你的Kafka Connect集群是分布式的，每个主机都运行着Connect服务，建议改成访问当前主机的服务，把URL改成http://{{inventory_hostname}}:8083/connectors/{{item.name}}/status，这样每个主机检查自己本地的连接器状态，更符合分布式部署的逻辑。

备注：内容来源于stack exchange，提问作者grizzo