寻求.NET环境下Kafka客户端单元测试的编写协助

Confluent的.NET Kafka客户端确实没有官方提供Mock生产者/消费者，但可以通过以下几种方案解决单元测试需求：

1. 抽象接口+自定义Mock实现 #

这是最灵活可控的方案，核心思路是对Confluent的生产者/消费者进行接口封装，然后在测试中使用自定义的Mock实现替代真实客户端。

步骤示例： #

• 定义抽象接口，包装核心操作：

public interface IKafkaProducer<TKey, TValue>
{
    Task ProduceAsync(string topic, Message<TKey, TValue> message, CancellationToken cancellationToken = default);
    void Flush(TimeSpan timeout);
}

public interface IKafkaConsumer<TKey, TValue>
{
    ConsumeResult<TKey, TValue> Consume(TimeSpan timeout);
    void Subscribe(string topic);
}

• 实现真实的Confluent客户端包装类：

public class ConfluentKafkaProducer<TKey, TValue> : IKafkaProducer<TKey, TValue>
{
    private readonly IProducer<TKey, TValue> _producer;

    public ConfluentKafkaProducer(IProducer<TKey, TValue> producer)
    {
        _producer = producer;
    }

    public Task ProduceAsync(string topic, Message<TKey, TValue> message, CancellationToken cancellationToken = default)
    {
        return _producer.ProduceAsync(topic, message, cancellationToken);
    }

    public void Flush(TimeSpan timeout)
    {
        _producer.Flush(timeout);
    }
}

• 编写Mock实现，用于测试验证：

public class MockKafkaProducer<TKey, TValue> : IKafkaProducer<TKey, TValue>
{
    public List<(string Topic, Message<TKey, TValue> Message)> ProducedMessages { get; } = new();

    public Task ProduceAsync(string topic, Message<TKey, TValue> message, CancellationToken cancellationToken = default)
    {
        ProducedMessages.Add((topic, message));
        return Task.CompletedTask;
    }

    public void Flush(TimeSpan timeout)
    {
        // 空实现或按需处理
    }
}

• 在测试中使用Mock：

[Fact]
public void WhenProducingMessage_ShouldTrackMessage()
{
    var mockProducer = new MockKafkaProducer<string, string>();
    var service = new YourService(mockProducer);

    service.SendMessage("test-topic", "key", "value");

    Assert.Single(mockProducer.ProducedMessages);
    Assert.Equal("test-topic", mockProducer.ProducedMessages[0].Topic);
    Assert.Equal("value", mockProducer.ProducedMessages[0].Message.Value);
}

2. 使用Moq模拟Confluent客户端 #

如果不想自定义接口，可以直接用Moq来模拟Confluent提供的IProducer<TKey, TValue>和IConsumer<TKey, TValue>接口（需确保使用的Confluent版本已提供这些接口）：

[Fact]
public async Task ProduceMessage_ShouldCallProduceAsync()
{
    var mockProducer = new Mock<IProducer<string, string>>();
    var service = new YourService(mockProducer.Object);

    await service.SendMessage("test-topic", "key", "value");

    mockProducer.Verify(p => p.ProduceAsync(
        "test-topic",
        It.Is<Message<string, string>>(m => m.Value == "value"),
        It.IsAny<CancellationToken>()),
        Times.Once);
}

3. 嵌入式Kafka实例（集成测试场景） #

如果需要更接近真实环境的测试，可以使用嵌入式Kafka，借助社区库在测试生命周期内启动临时Kafka集群：

public class KafkaIntegrationTest : IAsyncLifetime
{
    private IEmbeddedKafka _embeddedKafka;

    public async Task InitializeAsync()
    {
        _embeddedKafka = await EmbeddedKafka.StartAsync(new EmbeddedKafkaOptions
        {
            KafkaPort = 9092,
            ZooKeeperPort = 2181
        });
    }

    [Fact]
    public async Task ProduceAndConsume_ShouldWork()
    {
        var producerConfig = new ProducerConfig { BootstrapServers = "localhost:9092" };
        using var producer = new ProducerBuilder<string, string>(producerConfig).Build();

        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092",
            GroupId = "test-group",
            AutoOffsetReset = AutoOffsetReset.Earliest
        };
        using var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
        consumer.Subscribe("test-topic");

        await producer.ProduceAsync("test-topic", new Message<string, string> { Value = "test-value" });
        producer.Flush(TimeSpan.FromSeconds(1));

        var consumeResult = consumer.Consume(TimeSpan.FromSeconds(1));
        Assert.Equal("test-value", consumeResult.Message.Value);
    }

    public async Task DisposeAsync()
    {
        await _embeddedKafka.StopAsync();
    }
}

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