使用MassTransit Kafka传输时如何完全禁用librdkafka原生日志？

在使用MassTransit Kafka传输时，librdkafka的原生日志会直接输出到stderr，且不经过Serilog处理，示例日志如下：

%3|1770535125.181|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Unknown error (after 2028ms in state CONNECT)
%3|1770535125.181|ERROR|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]: 1/1 brokers are down
%3|1770535125.181|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Unknown error (after 2028ms in state CONNECT)

环境信息 #

• MassTransit：8.5.7
• Confluent.Kafka（隐式依赖）：2.12.0
• librdkafka.redist（隐式依赖）：2.12.0
• 操作系统：Windows / Docker（表现一致）

方案1：完全禁用librdkafka原生日志 #

librdkafka的日志行为由Confluent.Kafka配置项控制，你可以在MassTransit的Kafka配置中，为生产者和消费者设置日志级别为None，直接阻止所有日志输出到stderr：

services.AddMassTransit(cfg =>
{
    // 配置消费者、消息定义等逻辑

    cfg.UsingKafka((context, kafkaCfg) =>
    {
        kafkaCfg.Host("localhost:9092");

        // 配置生产者日志级别
        kafkaCfg.ProducerConfig(p =>
        {
            p.LogLevel = Confluent.Kafka.LogLevel.None;
            // 也可直接使用原始librdkafka配置项
            // p.Set("log.level", "none");
        });

        // 配置消费者日志级别
        kafkaCfg.ConsumerConfig(c =>
        {
            c.LogLevel = Confluent.Kafka.LogLevel.None;
        });

        // 绑定端点、配置消息拓扑等逻辑
    });
});

方案2：将librdkafka日志接入Serilog #

通过设置Confluent.Kafka的LogHandler回调，把librdkafka的日志事件转发到Serilog，同时会自动抑制默认的stderr输出：

services.AddMassTransit(cfg =>
{
    // 配置消费者、消息定义等逻辑

    cfg.UsingKafka((context, kafkaCfg) =>
    {
        kafkaCfg.Host("localhost:9092");

        // 生产者日志接入Serilog
        kafkaCfg.ProducerConfig(p =>
        {
            p.LogHandler = (producer, logEvent) =>
            {
                var serilogLevel = logEvent.Level switch
                {
                    Confluent.Kafka.LogLevel.Debug => Serilog.Events.LogEventLevel.Debug,
                    Confluent.Kafka.LogLevel.Info => Serilog.Events.LogEventLevel.Information,
                    Confluent.Kafka.LogLevel.Warning => Serilog.Events.LogEventLevel.Warning,
                    Confluent.Kafka.LogLevel.Error => Serilog.Events.LogEventLevel.Error,
                    Confluent.Kafka.LogLevel.Critical => Serilog.Events.LogEventLevel.Fatal,
                    _ => Serilog.Events.LogEventLevel.Verbose
                };

                Log.Write(serilogLevel, "[librdkafka {ProducerName}] {Message}", producer.Name, logEvent.Message);
            };
        });

        // 消费者日志接入Serilog
        kafkaCfg.ConsumerConfig(c =>
        {
            c.LogHandler = (consumer, logEvent) =>
            {
                var serilogLevel = logEvent.Level switch
                {
                    Confluent.Kafka.LogLevel.Debug => Serilog.Events.LogEventLevel.Debug,
                    Confluent.Kafka.LogLevel.Info => Serilog.Events.LogEventLevel.Information,
                    Confluent.Kafka.LogLevel.Warning => Serilog.Events.LogEventLevel.Warning,
                    Confluent.Kafka.LogLevel.Error => Serilog.Events.LogEventLevel.Error,
                    Confluent.Kafka.LogLevel.Critical => Serilog.Events.LogEventLevel.Fatal,
                    _ => Serilog.Events.LogEventLevel.Verbose
                };

                Log.Write(serilogLevel, "[librdkafka {ConsumerName}] {Message}", consumer.Name, logEvent.Message);
            };
        });

        // 绑定端点、配置消息拓扑等逻辑
    });
});

注意事项 #

• 需同时配置生产者和消费者的日志设置，避免遗漏某一方的日志输出。
• 如果仍有调试日志残留，可在生产者/消费者配置中添加p.Set("debug", "")，禁用librdkafka的调试日志模块。

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