如何使用IAM角色为Flink的OpenSearch-2连接器实现认证（对接AWS托管OpenSearch）

1. 依赖准备 #

确保你的Flink项目中包含以下必要依赖（以Maven为例）：

<!-- Flink OpenSearch 连接器 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-opensearch2_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>
<!-- AWS SDK for OpenSearch IAM 认证支持 -->
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>opensearch</artifactId>
    <version>2.20.0</version>
</dependency>
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>auth</artifactId>
    <version>2.20.0</version>
</dependency>

2. 连接器核心配置 #

在Flink的Sink配置中，指定IAM认证相关参数，替代传统的用户名密码：

import org.apache.flink.connector.opensearch.sink.OpensearchSink;
import org.apache.flink.connector.opensearch.sink.OpensearchSinkBuilder;
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
import org.opensearch.client.RestClientBuilder;
import software.amazon.opensearch.client.http.AwsSdk2HttpInterceptor;

// 构建OpenSearch客户端配置
RestClientBuilder restClientBuilder = RestClient.builder(new HttpHost("your-opensearch-domain-endpoint", 443, "https"));

// 配置IAM认证拦截器
restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> {
    httpClientBuilder.addInterceptorLast(new AwsSdk2HttpInterceptor(
        DefaultCredentialsProvider.create(),
        "us-east-1" // 替换为你的AWS区域
    ));
    return httpClientBuilder;
});

// 构建OpensearchSink
OpensearchSink<String> sink = new OpensearchSinkBuilder<String>()
    .setRestClientBuilder(restClientBuilder)
    .setIndexer((element, context, indexer) -> {
        // 自定义索引写入逻辑
        indexer.add(
            IndexRequest.of(i -> i.index("target-index").document(element))
        );
    })
    .build();

// 将Sink绑定到数据流
stream.sinkTo(sink);

关键配置说明：

• AwsSdk2HttpInterceptor：自动生成IAM签名，替代用户名密码认证逻辑
• DefaultCredentialsProvider：自动从环境变量、EC2实例角色、EKS Pod角色等渠道获取IAM凭证
• 若需指定特定IAM角色，可替换为AssumeRoleCredentialsProvider：

  AssumeRoleCredentialsProvider credentialsProvider = AssumeRoleCredentialsProvider.builder()
      .roleArn("arn:aws:iam::123456789012:role/your-flink-opensearch-role")
      .build();
  

3. IAM角色权限配置 #

确保Flink使用的IAM角色拥有访问目标OpenSearch域的权限，示例IAM策略：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "es:ESHttpPut",
                "es:ESHttpPost",
                "es:ESHttpDelete"
            ],
            "Resource": "arn:aws:es:us-east-1:123456789012:domain/your-opensearch-domain/*"
        }
    ]
}

4. 验证配置 #

• 启动Flink任务，查看日志确认无认证失败报错
• 检查OpenSearch目标索引，确认数据成功写入

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