Redpanda Connect嵌套map_each构建动态SQS输入的报错解决问询

问题背景 #

我尝试用Redpanda Connect的模板功能，通过动态参数构建SQS输入，目标是生成以下格式的SQS URL：

https://sqs.eu-west-1.amazonaws.com/123456789012/tr-dev-pip-myevent1
https://sqs.eu-west-1.amazonaws.com/123456789012/tr-dev-pip-myevent2
https://sqs.eu-west-1.amazonaws.com/123456789012/de-dev-pip-myevent1
https://sqs.eu-west-1.amazonaws.com/123456789012/de-dev-pip-myevent2

但编写嵌套map_each代码时，出现了语法解析错误：

aws_sqs_list.yaml(40,15) required: expected query, got: 
})

aws_sqs_list.yaml(1,1) parse mapping: line 13 char 5: required: expected query
   |
13 |   })
   |     ^---

我的模板代码如下：

name: aws_sqs_list
type: input
summary: Generates a list of SQS inputs based on dynamic parameters.

fields:
  - name: region
    type: string
    description: The AWS region for the SQS queues.
    default: eu-west-1
  - name: account_id
    type: string
    description: The AWS account ID.
  - name: cluster_name
    type: string
    description: The cluster identifier part of the queue name.
  - name: countries
    type: string
    kind: list
    description: A list of country codes to generate queues for.
  - name: event_names
    type: string
    kind: list
    description: A list of event names to generate queues for.

mapping: |
  root.broker.inputs = this.countries.map_each(country -> {
    this.event_names.map_each(event_name -> {
      "aws_sqs": {
        "url": "https://sqs.%s.amazonaws.com/%s/%s-%s-pip-%s".format(
          this.region,
          this.account_id,
          this.cluster_name,
          country,
          event_name
        ),
        "region": this.region
      }
    })
  })

解决方案 #

报错原因是嵌套map_each返回的是嵌套列表（每个国家对应一个事件列表），不符合Redpanda Connect对broker.inputs的扁平列表要求，同时语法上存在对象返回格式问题。

修正后的mapping代码如下：

mapping: |
  root.broker.inputs = this.countries.map_each(country -> {
    this.event_names.map_each(event_name -> {
      {
        "aws_sqs": {
          "url": "https://sqs.%s.amazonaws.com/%s/%s-%s-pip-%s".format(
            this.region,
            this.account_id,
            this.cluster_name,
            country,
            event_name
          ),
          "region": this.region
        }
      }
    })
  }).flatten()

关键修改点： #

• 在内部map_each的返回对象外层添加额外{}，确保返回的是单个完整对象而非直接的键值对，修复语法解析问题
• 在最外层map_each后调用.flatten()，将嵌套的二维列表展开为一维列表，匹配broker.inputs的格式要求

修改后即可正确生成目标格式的扁平SQS输入列表，语法错误也会消除。

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