如何通过托管服务身份使用Confluent Kafka .NET API连接Event Hub？

刚好之前做过类似的实现，用Confluent Kafka .NET客户端通过托管服务身份（MSI）连接Event Hub是完全可行的，核心是借助Azure的身份认证库来生成Kafka需要的OAuthBearer令牌，下面一步步给你讲清楚：

Event Hub的Kafka兼容端点支持SASL OAuthBearer认证机制，而Azure的MSI身份可以通过Azure.Identity库获取对应的TokenCredential，我们只需要把这个凭证转换成Kafka客户端能识别的令牌提供者，就能完成MSI身份的认证。

首先得把依赖包装好，直接在NuGet包管理器里搜索安装：

• Confluent.Kafka：Confluent官方的.NET Kafka客户端
• Azure.Identity：用于获取MSI或其他Azure身份的令牌
• System.Text.Json（可选，一般.NET项目都会自带）

我们需要自定义一个类，实现Confluent.Kafka提供的IOauthBearerTokenProvider接口，内部用Azure的TokenCredential来获取Event Hub的访问令牌：

using Confluent.Kafka;
using Azure.Identity;
using System.Threading.Tasks;

public class MsiKafkaTokenProvider : IOauthBearerTokenProvider
{
    private readonly TokenCredential _tokenCredential;
    // Event Hub的默认权限范围，固定值不用改
    private const string EventHubAuthScope = "https://eventhubs.azure.net/.default";

    public MsiKafkaTokenProvider()
    {
        // 使用DefaultAzureCredential会自动适配环境：
        // - 本地开发时用Azure CLI/Visual Studio登录的身份
        // - 托管环境（VM/AKS/App Service）自动用MSI
        _tokenCredential = new DefaultAzureCredential();
    }

    // 如果用的是用户分配的MSI，就用这个构造函数
    public MsiKafkaTokenProvider(string userAssignedMsiClientId)
    {
        _tokenCredential = new DefaultAzureCredential(new DefaultAzureCredentialOptions
        {
            ManagedIdentityClientId = userAssignedMsiClientId
        });
    }

    public async Task<OauthBearerToken> GetToken(OauthBearerTokenProviderConfig config)
    {
        // 获取Event Hub的访问令牌
        var tokenResult = await _tokenCredential.GetTokenAsync(
            new Azure.Core.TokenRequestContext(new[] { EventHubAuthScope })
        );

        // 转换成Kafka需要的令牌格式
        return new OauthBearerToken
        {
            Token = tokenResult.Token,
            ExpirationTimestamp = tokenResult.ExpiresOn.ToUnixTimeMilliseconds()
        };
    }
}

接下来就可以把这个令牌提供者配置到生产者或消费者里了：

生产者配置示例 #

var producerConfig = new ProducerConfig
{
    // 替换成你的Event Hub命名空间的Kafka端点
    BootstrapServers = "your-eventhub-namespace.servicebus.windows.net:9093",
    ClientId = "msi-kafka-producer",
    // 指定SASL认证机制为OAuthBearer
    SaslMechanism = SaslMechanism.OAuthBearer,
    SecurityProtocol = SecurityProtocol.SaslSsl,
    // 绑定我们自定义的令牌提供者
    OauthBearerTokenProvider = new MsiKafkaTokenProvider()
};

// 初始化生产者并发送消息
using var producer = new ProducerBuilder<Null, string>(producerConfig).Build();
var sendResult = await producer.ProduceAsync("your-eventhub-name", new Message<Null, string> { Value = "这条消息是用MSI身份发送的！" });

Console.WriteLine($"消息发送成功，偏移量：{sendResult.Offset}");

消费者配置示例 #

var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "your-eventhub-namespace.servicebus.windows.net:9093",
    GroupId = "msi-kafka-consumer-group",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    ClientId = "msi-kafka-consumer",
    SaslMechanism = SaslMechanism.OAuthBearer,
    SecurityProtocol = SecurityProtocol.SaslSsl,
    OauthBearerTokenProvider = new MsiKafkaTokenProvider()
};

// 初始化消费者并订阅消息
using var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build();
consumer.Subscribe("your-eventhub-name");

Console.WriteLine("开始监听消息...");
while (true)
{
    var consumeResult = consumer.Consume();
    Console.WriteLine($"收到消息：{consumeResult.Message.Value}，偏移量：{consumeResult.Offset}");
}

• 权限配置：一定要给你的MSI身份分配对应的Event Hub权限——生产者需要Azure Event Hub Data Sender角色，消费者需要Azure Event Hub Data Receiver角色，直接在Event Hub命名空间或具体的Event Hub资源上分配即可。
• 本地调试：用DefaultAzureCredential的话，本地开发时会自动用你Azure CLI或者Visual Studio登录的身份，不需要额外配置，调试起来很方便。
• 令牌自动刷新：Confluent的Kafka客户端会自动在令牌过期前调用GetToken方法刷新，不需要我们手动处理令牌过期问题。

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