.NET 6下如何用NUnit测试Confluent.Kafka生产者与消费者客户端

一、解决Moq注入问题：通过抽象接口解耦 #

Confluent.Kafka的ConsumerBuilder和IConsumer<TKey, TValue>直接Mock容易出现注入问题，核心原因是业务代码可能直接依赖了具体实现类。解决思路是将业务逻辑与Kafka客户端的具体实现解耦：

1. 定义业务抽象接口 #

先针对Kafka消费者、生产者定义业务层面的抽象接口，只保留你实际用到的方法：

public interface IKafkaConsumer<TKey, TValue>
{
    ConsumeResult<TKey, TValue> Consume(CancellationToken cancellationToken);
    void Subscribe(string topic);
    void Commit(ConsumeResult<TKey, TValue> result);
    // 按需添加其他方法
}

public interface IKafkaProducer<TKey, TValue>
{
    Task ProduceAsync(string topic, Message<TKey, TValue> message, CancellationToken cancellationToken);
    void Flush(TimeSpan timeout);
    // 按需添加其他方法
}

2. 实现Confluent客户端的包装类 #

把Confluent官方的Consumer/Producer包装到上述接口的实现类中，隔离具体依赖：

public class KafkaConsumer<TKey, TValue> : IKafkaConsumer<TKey, TValue>, IDisposable
{
    private readonly IConsumer<TKey, TValue> _innerConsumer;

    public KafkaConsumer(ConsumerConfig config)
    {
        _innerConsumer = new ConsumerBuilder<TKey, TValue>(config).Build();
    }

    public ConsumeResult<TKey, TValue> Consume(CancellationToken cancellationToken)
    {
        return _innerConsumer.Consume(cancellationToken);
    }

    public void Subscribe(string topic)
    {
        _innerConsumer.Subscribe(topic);
    }

    public void Commit(ConsumeResult<TKey, TValue> result)
    {
        _innerConsumer.Commit(result);
    }

    public void Dispose()
    {
        _innerConsumer.Dispose();
    }
}

3. 业务代码依赖抽象接口 #

你的业务逻辑类不再直接依赖ConsumerBuilder或IConsumer，而是注入上述抽象接口：

public class OrderProcessor
{
    private readonly IKafkaConsumer<string, Order> _consumer;

    public OrderProcessor(IKafkaConsumer<string, Order> consumer)
    {
        _consumer = consumer;
    }

    public async Task ProcessNextOrder(CancellationToken ct)
    {
        var consumeResult = _consumer.Consume(ct);
        var order = consumeResult.Message.Value;
        // 处理订单逻辑
        _consumer.Commit(consumeResult);
    }
}

4. 用Moq编写NUnit测试 #

现在可以轻松Mock抽象接口，无需再处理ConsumerBuilder的初始化问题：

[TestFixture]
public class OrderProcessorTests
{
    [Test]
    public void ProcessNextOrder_ShouldCommitAfterProcessing()
    {
        // Arrange
        var mockConsumer = new Mock<IKafkaConsumer<string, Order>>();
        var testOrder = new Order { Id = "123", Amount = 99.9m };
        var mockConsumeResult = new ConsumeResult<string, Order>
        {
            Message = new Message<string, Order> { Value = testOrder },
            TopicPartitionOffset = new TopicPartitionOffset("orders", 0, Offset.At(10))
        };
        
        mockConsumer.Setup(c => c.Consume(It.IsAny<CancellationToken>()))
            .Returns(mockConsumeResult);

        var processor = new OrderProcessor(mockConsumer.Object);

        // Act
        processor.ProcessNextOrder(CancellationToken.None).Wait();

        // Assert
        mockConsumer.Verify(c => c.Commit(mockConsumeResult), Times.Once);
    }
}

二、使用Confluent官方Mock类（需自行引入） #

Confluent.Kafka的GitHub仓库中提供了测试专用的MockConsumer和MockProducer类，但未打包到NuGet包中。你可以直接从对应版本的测试代码中复制这些类到你的测试项目：

1. 找到Confluent.Kafka v1.9.3版本的测试代码（路径：tests/Confluent.Kafka.Tests/Mock）
2. 复制MockConsumer.cs和MockProducer.cs到你的NUnit测试项目
3. 直接使用这些类模拟真实的Kafka交互，适合测试分区分配、偏移量管理等复杂逻辑：

[TestFixture]
public class KafkaConsumerIntegrationTests
{
    [Test]
    public void Consume_ShouldReturnMockedRecordsInOrder()
    {
        var mockConsumer = new MockConsumer<string, string>(new ConsumerConfig());
        // 分配分区
        mockConsumer.Assign(new List<TopicPartitionOffset> { new TopicPartitionOffset("test-topic", 0, Offset.Beginning) });
        // 添加模拟消息
        mockConsumer.AddRecord(new Message<string, string> { Value = "first message" });
        mockConsumer.AddRecord(new Message<string, string> { Value = "second message" });

        // 消费第一条
        var result1 = mockConsumer.Consume(CancellationToken.None);
        Assert.That(result1.Message.Value, Is.EqualTo("first message"));
        // 消费第二条
        var result2 = mockConsumer.Consume(CancellationToken.None);
        Assert.That(result2.Message.Value, Is.EqualTo("second message"));
    }
}

三、生产者测试的同类方案 #

抽象接口+Moq方式 #

[Test]
public async Task ProduceOrder_ShouldCallProduceAsyncWithCorrectTopic()
{
    var mockProducer = new Mock<IKafkaProducer<string, Order>>();
    var testOrder = new Order { Id = "456", Amount = 199.9m };
    var processor = new OrderProducer(mockProducer.Object);

    await processor.SendOrder("orders", testOrder);

    mockProducer.Verify(p => p.ProduceAsync(
        "orders", 
        It.Is<Message<string, Order>>(m => m.Value.Id == "456"), 
        It.IsAny<CancellationToken>()), 
        Times.Once);
}

MockProducer方式 #

[Test]
public async Task ProduceAsync_ShouldRecordAllMessages()
{
    var mockProducer = new MockProducer<string, Order>(new ProducerConfig());
    var producer = new KafkaProducer<string, Order>(new ProducerConfig());

    await mockProducer.ProduceAsync("orders", new Message<string, Order> { Value = new Order { Id = "789" } });
    await mockProducer.ProduceAsync("orders", new Message<string, Order> { Value = new Order { Id = "012" } });

    var producedMessages = mockProducer.LatestProduced;
    Assert.That(producedMessages.Count, Is.EqualTo(2));
    Assert.That(producedMessages.Select(m => m.Message.Value.Id), Contains.Item("789"));
}

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