Benthos YAML流水线HTTP错误捕获异常：如何保留错误信息至错误主题？

原配置的核心问题 #

1. branch 处理器默认不会将分支内的错误传播到主消息流，导致主消息不会被标记为 errored()，输出环节的switch无法匹配错误分支。
2. catch块内执行root = deleted()直接删除了消息，后续输出环节自然无法获取错误内容。
3. 错误信息未正确留存到元数据或消息体中，分支内的修改未传递到主消息上下文。

修改后的完整配置示例 #

http:
  address: 0.0.0.0:4190
  root_path: /benthos
input:
  kafka_franz:
    seed_brokers:
      - seed-broker-server:15555
    consumer_group: "consumer_group1"
    topic: in_topic

pipeline:
  processors:
    - branch:
        propagate_errors: true  # 关键：将分支错误传播到主消息流
        processors:
          - mapping: |
              root = { "input_param1": "value1" }.key_values()
          - http:
              url: https://api.sampleserver.com/firstapi
              verb: POST
              retries: 0  # 可选：禁用HTTP重试，直接触发错误捕获
          - catch:
              - mapping: |
                  # 方案1：将错误信息存入元数据，保留原始消息
                  meta error_details = error()
                  # 方案2：直接构造错误消息体（二选一即可）
                  # root = {
                  #   "original_input": json(),
                  #   "error": error(),
                  #   "kafka_key": meta("kafka_key")
                  # }
        # 其他业务流程代码

output:
  switch:
    cases:
      - check: errored()
        output:
          kafka_franz:
            seed_brokers: out_broker_server:16666
            topic: "error_topic"
            key: ${! meta("kafka_key") }
          processors:
            - log:
                level: INFO
                message: "Catch and send to error topic: ${! meta("error_details") }"
    - output:
        # 正常输出逻辑代码

关键修改说明 #

• propagate_errors: true：开启后，分支内的HTTP错误会标记主消息为errored()，确保输出环节的switch能匹配到错误分支。
• 移除root = deleted()：该操作会直接丢弃消息，必须删除才能让错误消息进入输出流程。
• 错误信息留存方式：
  • 存入元数据：适合需要保留原始消息内容，同时附加错误详情的场景，后续通过meta("error_details")读取。
  • 构造错误消息体：直接生成错误主题所需的结构化消息，包含原始输入和错误信息，无需依赖元数据。
• http.retries: 0：禁用HTTP请求重试，避免不必要的等待，直接进入错误捕获流程（按需配置）。

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