如何使用IAM认证实现C# .NET Kafka生产者连接AWS MSK？

Confluent.Kafka的.NET库本身没有内置AWS MSK的IAM认证支持，但可以通过SASL OAUTHBEARER机制结合AWS官方工具包来实现认证，具体方案如下：

1. 依赖准备 #

除了Confluent.Kafka，需要安装两个关键NuGet包：

• AWSSDK.Core：AWS基础SDK组件
• AWS.MSK.IAM：AWS官方提供的MSK IAM认证工具包，用于简化令牌生成流程

2. 完整代码实现 #

using Confluent.Kafka;
using Amazon;
using Amazon.MSK.IAM;
using Amazon.MSK.IAM.Model;

// 替换为你的实际配置信息
var awsRegion = RegionEndpoint.USEast1;
var mskClusterArn = "arn:aws:kafka:us-east-1:123456789012:cluster/your-cluster-id/xxxxxx";
var bootstrapServers = "broker-1.host:port,broker-2.host:port,broker-3.host:port";
var targetTopic = "your-target-topic";

// 1. 配置生产者核心参数
var producerConfig = new ProducerConfig
{
    SecurityProtocol = SecurityProtocol.SaslSsl,
    SaslMechanism = SaslMechanism.OauthBearer,
    ClientId = "table-producer",
    BootstrapServers = bootstrapServers,
    // 绑定自定义AWS MSK令牌提供器
    OAuthBearerTokenProvider = new AwsMskTokenProvider(mskClusterArn, awsRegion)
};

// 2. 初始化生产者并发送消息
using (var producer = new ProducerBuilder<Null, string>(producerConfig).Build())
{
    try
    {
        var message = new Message<Null, string> { Value = "test message from IAM-authenticated producer" };
        var result = await producer.ProduceAsync(targetTopic, message);
        Console.WriteLine($"消息发送成功：分区 {result.Partition}，偏移量 {result.Offset}");
    }
    catch (ProduceException<Null, string> ex)
    {
        Console.WriteLine($"发送失败：{ex.Error.Reason}");
    }
}

// 自定义AWS MSK令牌提供器，实现Confluent的IOAuthBearerTokenProvider接口
public class AwsMskTokenProvider : IOAuthBearerTokenProvider
{
    private readonly string _clusterArn;
    private readonly IAmazonMSKIAM _mskIamClient;

    public AwsMskTokenProvider(string clusterArn, RegionEndpoint region)
    {
        _clusterArn = clusterArn;
        _mskIamClient = new AmazonMSKIAMClient(region);
    }

    public async Task<OAuthBearerToken> GetToken(OAuthBearerTokenProviderConfig config)
    {
        // 调用AWS官方API获取符合MSK要求的Bearer令牌
        var request = new GetBearerTokenRequest { ClusterArn = _clusterArn };
        var response = await _mskIamClient.GetBearerTokenAsync(request);

        return new OAuthBearerToken
        {
            Token = response.BearerToken,
            Expiration = DateTime.UtcNow.AddSeconds(response.ExpiresInSeconds)
        };
    }
}

3. 必要前置条件 #

• MSK集群已启用IAM认证：在AWS控制台的MSK集群配置中开启SASL/IAM认证选项
• IAM实体权限配置：确保运行代码的IAM用户/角色拥有kafka:DescribeCluster、kafka:WriteData以及kafka:GetBearerToken的权限
• 有效AWS凭证：本地开发可通过~/.aws/credentials配置凭证，服务器环境可使用EC2实例角色、ECS任务角色等自动获取凭证的方式

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