如何捕获并处理Kafka发送无效JSON时的MassTransit消费异常？

核心思路 #

在Kafka消费循环里精准捕获反序列化异常，记录关键错误信息后手动提交偏移量，直接跳过这条无效消息。

具体实现步骤 #

• 捕获目标异常：重点抓ConsumeException，并判断其内部异常是否为JsonException——这就是JSON反序列化失败的典型场景。
• 记录详细日志：把异常堆栈、消息的主题/分区/偏移量、原始消息内容都记下来，方便后续定位问题根源。
• 提交偏移量跳过消息：捕获异常后手动提交当前消息的偏移量，确保下次消费不会重复处理这条无效数据。

代码示例（.NET环境） #

using Confluent.Kafka;
using System.Text.Json;
using System.Text;
using Microsoft.Extensions.Logging;

// 假设已初始化消费者和日志实例
var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "your-kafka-broker",
    GroupId = "your-consumer-group",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false // 禁用自动提交，改为手动控制
};

var consumer = new ConsumerBuilder<Ignore, byte[]>(consumerConfig).Build();
var logger = LoggerFactory.Create(builder => builder.AddConsole()).CreateLogger<ConsumerWorker>();

consumer.Subscribe("your-target-topic");

try
{
    while (true)
    {
        try
        {
            var consumeResult = consumer.Consume(TimeSpan.FromSeconds(1));
            if (consumeResult == null) continue;

            // 尝试反序列化消息
            var message = JsonSerializer.Deserialize<YourMessageModel>(consumeResult.Message.Value);
            
            // 正常业务处理逻辑
            HandleValidMessage(message);
            
            // 处理完成后提交偏移量
            consumer.Commit(consumeResult);
        }
        catch (ConsumeException ex) when (ex.InnerException is JsonException jsonEx)
        {
            // 记录完整错误信息
            var rawMessage = consumeResult?.Message.Value != null 
                ? Encoding.UTF8.GetString(consumeResult.Message.Value) 
                : "空消息内容";
            
            logger.LogError(jsonEx, "JSON反序列化失败 | 主题：{Topic} | 分区：{Partition} | 偏移量：{Offset} | 原始内容：{RawContent}",
                consumeResult.Topic, consumeResult.Partition, consumeResult.Offset, rawMessage);
            
            // 提交偏移量，跳过这条无效消息
            consumer.Commit(consumeResult);
        }
        catch (Exception ex)
        {
            // 处理其他未知异常
            logger.LogError(ex, "消费消息时发生未预期错误");
            // 非反序列化异常可根据业务决定是否重试或提交偏移量
        }
    }
}
finally
{
    consumer.Close();
}

注意事项 #

• 建议禁用自动偏移量提交，改为手动提交，这样能精准控制哪些消息的偏移量被确认。
• 如果业务需要保留无效消息用于后续排查，可以把消息转发到死信队列（DLQ），而不是直接跳过。
• 日志要包含足够的上下文信息，比如消息的元数据和原始内容，不然很难定位问题。

内容的提问来源于stack exchange，提问作者Mikhail Sh.