如何在AWS Lambda中基于Kafka消息头值过滤消息？

问题背景 #

我用自托管Kafka触发器触发AWS Lambda，收到的消息格式如下：

{
   "eventSource":"aws:kafka",
   "eventSourceArn":"arn:aws:kafka:sa-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",
   "bootstrapServers":"b-2.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092,b-1.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092",
   "records":{
      "mytopic-0":[
         {
            "topic":"mytopic",
            "partition":0,
            "offset":15,
            "timestamp":1545084650987,
            "timestampType":"CREATE_TIME",
            "key":"abcDEFghiJKLmnoPQRstuVWXyz1234==",
            "value":"SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
            "headers":[
               {
                  "headerKey":[
                     104,
                     101,
                     97,
                     100,
                     101,
                     114,
                     86,
                     97,
                     108,
                     117,
                     101
                  ]
               }
            ]
         }
      ]
   }
}

这里headerKey对应的整数数组解码后是字符串headerValue（用bytes(h).decode()就能转）。我想给触发器加过滤规则，只让headerKey等于headerValue的消息触发Lambda，减少无效调用。但AWS文档里的过滤规则都是针对字符串的，不知道怎么处理这种整数数组格式的消息头，求解决办法。

解决方案 #

Lambda的事件过滤规则其实支持直接匹配整数数组，按下面的步骤来就行：

• 明确匹配逻辑：Lambda过滤规则里的数组匹配是精确匹配元素的顺序和每个值，所以你直接把headerValue对应的ASCII整数数组写到规则里就行，不用提前解码。
• 配置过滤规则：
  在Kafka触发器的事件过滤规则中，添加以下JSON规则：

  {
    "records": {
      "mytopic-0": [
        {
          "headers": [
            {
              "headerKey": [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]
            }
          ]
        }
      ]
    }
  }
  

  如果你的消息可能有多个header，或者headerKey在headers数组里的位置不固定，可以用通配符匹配任意位置的header对象；要是想匹配所有topic分区，也可以把mytopic-0换成*：

  {
    "records": {
      "*": [
        {
          "headers": [
            {
              "headerKey": [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]
            }
          ]
        }
      ]
    }
  }
  

• 验证规则有效性：配置完后，发一条带headerKey: headerValue的消息到Kafka，看看Lambda会不会被触发；再发一条不匹配的消息，确认Lambda不会运行，这样就能验证规则是否生效。

注意事项 #

• Lambda的事件过滤支持嵌套结构和数组精确匹配，不需要提前处理整数数组，直接用原始的ASCII码数组就能匹配。
• 如果你的header值用了其他编码，得确保过滤规则里的整数数组和消息头的原始字节数组完全一致，不然匹配不上。

内容的提问来源于stack exchange，提问作者Eugene W.