如何为KafkaFlow配置证书认证的SSL连接

背景场景 #

我有一个基于.NET 5的C#项目，使用Confluent包成功连接Kafka，认证配置代码如下：

public ConsumerConfig Adapt(KafkaReaderConfiguration config)
{
    var cfg = new ConsumerConfig
    {
        GroupId = config.GroupId,
        BootstrapServers = config.Servers,
        AutoOffsetReset = config.StartPositionType,
        QueuedMinMessages = config.MinimumQueuedMessages,
        SslCertificateLocation = config.MutualAuthentication.CertificateFile,
        SslKeyLocation = config.MutualAuthentication.CertificateKey,
        SecurityProtocol = SecurityProtocol.Ssl,
        SslCaLocation = config.MutualAuthentication.CertificatePath,
        EnableSslCertificateVerification = false
    };
    return cfg;
}

现在我要在.NET 9中重构该项目并使用KafkaFlow框架（以利用其多线程、批处理等特性），但找不到KafkaFlow使用证书连接Kafka的相关文档。我当前的KafkaFlow代码如下：

builder.Services.AddKafka(kafka => kafka
    .UseMicrosoftLog()
    .AddCluster(cluster => cluster
        .WithBrokers(kafkaConfig.Brokers) 
        .AddConsumer(consumer => consumer
            .Topic(kafkaConfig.Topic)
            .WithGroupId(kafkaConfig.GroupId) 
            .WithBufferSize(kafkaConfig.BufferSize) 
            .WithWorkersCount(kafkaConfig.WorkersCount)
            .AddMiddlewares(middlewares => middlewares
                .AddDeserializer<JsonCoreDeserializer>() 
                .AddBatching(kafkaConfig.BatchSize, kafkaConfig.BatchTimeout) 
                .AddTypedHandlers(handlers => handlers
                    .AddHandler<AlertMessageHandler>() 
                )
            )
        )
    )
);

我原本期望能有类似如下的配置方式，但WithSecurityInformation方法并不存在：

builder.Services.AddKafka(kafka => kafka
    .UseMicrosoftLog()
    .AddCluster(cluster => cluster
        .WithBrokers(kafkaConfig.Brokers)  
        .AddConsumer(consumer => consumer
            .Topic(kafkaConfig.Topic)  
            // ... 其他配置
            .WithSecurityInformation(new SecurityInformation
            {
                SslCertificateLocation = kafkaConfig.MutualAuthentication.CertificateFile,
                SslKeyLocation = kafkaConfig.MutualAuthentication.CertificateKey,
                SecurityProtocol = SecurityProtocol.Ssl,
                SslCaLocation = kafkaConfig.MutualAuthentication.CertificatePath,
                EnableSslCertificateVerification = false
            })
        )
    )
);

请问该如何设置KafkaFlow以使用SSL证书进行身份认证？

解决方案 #

KafkaFlow底层依赖Confluent的.NET Kafka客户端，因此你可以直接复用原来的ConsumerConfig配置，通过KafkaFlow提供的配置扩展方法注入：

方式1：在集群级别配置（适用于消费者、生产者共用相同SSL配置的场景） #

builder.Services.AddKafka(kafka => kafka
    .UseMicrosoftLog()
    .AddCluster(cluster => cluster
        .WithBrokers(kafkaConfig.Brokers)
        // 全局配置SSL参数，集群内的消费者、生产者都会继承该配置
        .WithConfig(new ConsumerConfig
        {
            SecurityProtocol = SecurityProtocol.Ssl,
            SslCertificateLocation = kafkaConfig.MutualAuthentication.CertificateFile,
            SslKeyLocation = kafkaConfig.MutualAuthentication.CertificateKey,
            SslCaLocation = kafkaConfig.MutualAuthentication.CertificatePath,
            EnableSslCertificateVerification = false
        })
        .AddConsumer(consumer => consumer
            .Topic(kafkaConfig.Topic)
            .WithGroupId(kafkaConfig.GroupId)
            .WithBufferSize(kafkaConfig.BufferSize)
            .WithWorkersCount(kafkaConfig.WorkersCount)
            .AddMiddlewares(middlewares => middlewares
                .AddDeserializer<JsonCoreDeserializer>()
                .AddBatching(kafkaConfig.BatchSize, kafkaConfig.BatchTimeout)
                .AddTypedHandlers(handlers => handlers
                    .AddHandler<AlertMessageHandler>()
                )
            )
        )
    )
);

方式2：单独为消费者配置（适用于仅消费者需要特定SSL配置的场景） #

builder.Services.AddKafka(kafka => kafka
    .UseMicrosoftLog()
    .AddCluster(cluster => cluster
        .WithBrokers(kafkaConfig.Brokers)
        .AddConsumer(consumer => consumer
            .Topic(kafkaConfig.Topic)
            .WithGroupId(kafkaConfig.GroupId)
            .WithBufferSize(kafkaConfig.BufferSize)
            .WithWorkersCount(kafkaConfig.WorkersCount)
            // 仅为当前消费者配置SSL参数
            .WithConsumerConfig(new ConsumerConfig
            {
                SecurityProtocol = SecurityProtocol.Ssl,
                SslCertificateLocation = kafkaConfig.MutualAuthentication.CertificateFile,
                SslKeyLocation = kafkaConfig.MutualAuthentication.CertificateKey,
                SslCaLocation = kafkaConfig.MutualAuthentication.CertificatePath,
                EnableSslCertificateVerification = false
            })
            .AddMiddlewares(middlewares => middlewares
                .AddDeserializer<JsonCoreDeserializer>()
                .AddBatching(kafkaConfig.BatchSize, kafkaConfig.BatchTimeout)
                .AddTypedHandlers(handlers => handlers
                    .AddHandler<AlertMessageHandler>()
                )
            )
        )
    )
);

关键说明 #

• KafkaFlow并没有封装独立的SSL配置API，而是直接透传Confluent客户端的配置对象，因此你可以完全复用之前在Confluent中使用的所有SSL相关参数。
• 如果需要配置生产者的SSL参数，同样可以在集群级别用.WithConfig(new ProducerConfig(...))，或者在生产者配置中用.WithProducerConfig(...)方法。

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