无法修改消息Payload时，如何通过Kafka REST接入Azure ACR Webhook事件？

方案1：Azure Functions 中间转换层（低成本、轻量） #

这是最适配你场景的方案，Azure Functions 提供免费额度，低流量场景下几乎零成本。只需创建一个HTTP触发的函数，接收ACR Webhook的原始请求，将Payload包装成Kafka REST要求的格式后转发：

实现步骤： #

1. 在Azure门户创建HTTP触发的Azure Function（推荐Python或C#）
2. 编写函数逻辑：
   • 接收ACR发送的POST请求，提取原始Payload
   • 将原始Payload包装为 {"records": [{"value": 原始Payload}]} 格式
   • 向Kafka REST Proxy发送POST请求，设置正确的Content-Type头
3. 将ACR Webhook的目标地址设置为该Function的触发URL

Python示例代码： #

import requests
import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
    # 获取ACR原始Payload
    acr_payload = req.get_json()
    
    # 包装成Kafka REST要求的格式
    kafka_payload = {
        "records": [
            {"value": acr_payload}
        ]
    }
    
    # 发送到Kafka REST Proxy
    kafka_rest_uri = "你的Kafka REST Proxy地址"
    headers = {
        "Content-Type": "application/vnd.kafka.json.v2+json"
    }
    
    try:
        response = requests.post(kafka_rest_uri, json=kafka_payload, headers=headers)
        response.raise_for_status()
        return func.HttpResponse("转发成功", status_code=200)
    except Exception as e:
        return func.HttpResponse(f"转发失败: {str(e)}", status_code=500)

方案2：Azure API Management（APIM）无代码转换 #

如果不想编写代码，可以用APIM的入站策略自动转换请求格式，适合非开发人员操作：

配置步骤： #

1. 创建APIM实例（基础层足以应对低流量场景）
2. 在APIM中创建API，将后端服务地址设为你的Kafka REST Proxy地址
3. 为该API添加入站策略，自动包装原始请求Body：

<policies>
    <inbound>
        <base />
        <!-- 将原始Body包装为Kafka REST格式 -->
        <set-body>@{
            var originalBody = context.Request.Body.As<string>();
            return "{\"records\": [{\"value\": " + originalBody + "}]}";
        }</set-body>
        <!-- 强制设置正确的Content-Type -->
        <set-header name="Content-Type" exists-action="override">
            <value>application/vnd.kafka.json.v2+json</value>
        </set-header>
    </inbound>
    <backend>
        <base />
    </backend>
    <outbound>
        <base />
    </outbound>
    <on-error>
        <base />
    </on-error>
</policies>

4. 将ACR Webhook的目标地址设置为APIM的API调用地址

方案3：自定义轻量HTTP网关（灵活可控） #

如果具备开发能力，可以用FastAPI/Express.js编写简单的HTTP服务，部署在Azure App Service的免费层或低成本实例上，实现格式转换：

FastAPI示例（Python）： #

from fastapi import FastAPI, Request
import requests

app = FastAPI()

KAFKA_REST_URI = "你的Kafka REST Proxy地址"

@app.post("/acr-webhook-forward")
async def forward_acr_event(request: Request):
    acr_payload = await request.json()
    kafka_payload = {"records": [{"value": acr_payload}]}
    
    headers = {"Content-Type": "application/vnd.kafka.json.v2+json"}
    response = requests.post(KAFKA_REST_URI, json=kafka_payload, headers=headers)
    response.raise_for_status()
    return {"status": "success"}

无需Event Hub的Azure Webhook发Kafka的配置方式 #

上述三个方案均无需使用Event Hub，核心逻辑是通过中间转换层适配ACR Webhook的Payload格式与Kafka REST Proxy的要求。只需将ACR Webhook的目标地址指向中间层服务（Function/APIM/自定义网关），由中间层完成格式转换后转发至Kafka即可。

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