如何通过IAM角色创建Debezium ElasticsearchSinkConnector连接AWS ES实例（遇401错误）

我们尝试通过Debezium ElasticsearchSinkConnector连接Amazon Elasticsearch Service（ES）时，触发401未授权错误。此前未启用IAM认证时，可直接通过IAM角色连接ES实例；但启用IAM认证后，连接失败。本地无IAM认证的ES环境下，连接器可正常运行。虽已配置相关权限，但无法生成所需的access_key、secret_key和session_token，同时尝试用户名密码认证也未通过。

创建连接器的Curl命令如下：

curl --header 'Accept: application/json' \
--header 'Content-Type: application/json' \
--data '{
  "name": "elastic-sink-connect-v1",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "social.facebook_insight",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "connection.url": "https://search-replaceme.eu-west-1.es.amazonaws.com",
    "type.name": "_doc",
    "behavior.on.null.values": "delete",
    "elastic.security.protocol": "PLAINTEXT",
    "elastic.http.auth.type": "IAM", 
    "elastic.region": "us-west-1",
    "schema.ignore": "true",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "drop"
  }
}'

排查与解决步骤 #

1. 修正配置中的明显错误 #

• 区域不匹配：ES实例区域为eu-west-1（从connection.url可见），但配置里elastic.region设为us-west-1，这会导致IAM签名计算错误，需改为eu-west-1。
• 安全协议错误：启用IAM认证的AWS ES使用HTTPS协议，需将elastic.security.protocol改为SSL，而非PLAINTEXT。

修正后的核心配置片段：

"elastic.security.protocol": "SSL",
"elastic.region": "eu-west-1"

2. 确保IAM角色权限正确 #

给运行Kafka Connect的实例/服务（如EC2、ECS）绑定的IAM角色，添加以下ES权限策略：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "es:ESHttpPost",
        "es:ESHttpPut",
        "es:ESHttpDelete",
        "es:ESHttpGet"
      ],
      "Resource": "arn:aws:es:eu-west-1:你的AWS账号ID:domain/你的ES域名/*"
    }
  ]
}

确认该角色已正确关联到运行Connect的环境（如EC2实例的IAM角色、ECS任务执行角色）。

3. IAM凭证自动获取配置 #

• 若Connect运行在AWS托管服务（EC2、ECS、EKS）上，无需手动配置access_key/secret_key，连接器会自动从实例元数据服务（IMDS）获取临时凭证。
• 若运行在本地/非AWS环境，需设置环境变量AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY、AWS_SESSION_TOKEN（临时凭证需此参数），或配置AWS本地凭证文件（~/.aws/credentials），确保连接器能读取到这些凭证。

4. 检查连接器版本兼容性 #

确保使用的io.confluent.connect.elasticsearch.ElasticsearchSinkConnector版本支持IAM认证（建议使用Confluent Platform 5.5+版本），旧版本可能缺少IAM认证逻辑。

5. 用户名密码认证的正确配置（备选方案） #

若切换到用户名密码认证，修改以下配置：

"elastic.http.auth.type": "BASIC",
"elastic.http.auth.user": "你的ES用户名",
"elastic.http.auth.password": "你的ES密码",
"elastic.security.protocol": "SSL"

同时确保该用户已被授予ES的读写权限，且用户名密码无误。

6. 验证IAM权限有效性 #

通过AWS CLI测试ES访问权限，确认IAM凭证和权限正常：

aws es describe-elasticsearch-domain --domain-name 你的ES域名 --region eu-west-1

若CLI能正常返回结果，说明权限和凭证无问题，需排查连接器配置或运行环境。

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