使用TestContainers的Kafka集成测试随机失败问题求助

问题背景 #

使用TestContainers搭建Kafka环境做集成测试，测试包含生产/消费消息、数据库初始化逻辑。单独运行单个测试全过，但整套件运行时随机失败。已做的隔离措施：

• 每个测试用唯一Topic（通过Guid.NewGuid()后缀实现）
• 每个测试用全新命名的内存数据库
• 每个测试创建独立的Consumer（带唯一GroupId）和Producer
• Kafka容器原本是一次性初始化销毁，后来尝试每个测试重建容器，仍有部分失败

核心测试框架代码如下：

protected TDataContext _dataContext;
protected ReferenceDataContext _refContext;
protected IContainer _kafkaContainer;
protected IHost _host;
private Fixture _fixture;
private int _port;
private AO2KafkaConfiguration _kafkaConfiguration;
private IConsumer<Ignore, string> _consumer;
private IProducer<string, byte[]> _producer;
private Dictionary<string, string> _consumerTopics;
private Dictionary<string, string> _producerTopics;
private List<string> _topics => [.. _consumerTopics.Values, .. _producerTopics.Values];
protected string _bootstrapServer;
private IHostBuilder _hostBuilder;

protected abstract IHostBuilder GetHostBuilder();
protected abstract Task SetDataInDatabase();

[OneTimeSetUp]
public async Task OneTimeSetup()
{
    _fixture = new Fixture();
    _port = _fixture.Create<int>() % (5000 - 1000 + 1) + 1000;
    _bootstrapServer = GenerateBootstrapServer();
    await StartKafkaContainerAsync();
}

[OneTimeTearDown]
public async Task OneTimeTearDown()
{
    await _kafkaContainer.DisposeAsync();
}

[SetUp]
public async Task SetUp()
{
    _hostBuilder = GetHostBuilder();
    ConfigureKafkaConfiguration();
    _consumer = CreateConsumer();
    _producer = CreateProducer();
    InitializeDbsAsync();
    _host = _hostBuilder.Build();
    await CreateTopicsAsync();
}

[TearDown]
public async Task TearDown()
{
    _consumer.Dispose();
    _producer.Dispose();
    await _dataContext.DisposeAsync();
    await _refContext.DisposeAsync();
    _host.Dispose();
    await DeleteTopicsAsync();
}

// 核心消费/生产方法
protected ConsumeResult<Ignore, string> ConsumeTestMessage(string topicName, int timeoutMilliseconds, int retryRounds)
{
    _consumer.Subscribe(topicName);
    ConsumeResult<Ignore, string> consumeResult = null;
    retryRounds = 5;
    timeoutMilliseconds = 5000;
    var counter = 0;
    while (counter < retryRounds)
    {
        consumeResult = _consumer.Consume(TimeSpan.FromMilliseconds(timeoutMilliseconds));
        if (consumeResult is not null)
        {
            return consumeResult;
        }
        counter++;
        Task.Delay(timeoutMilliseconds);
    }
    return consumeResult;
}

protected async Task ProduceTestMessageAsync<T>(string publishTopicName, string jsonFixture)
{
    var message = await ReadMessageAsync<T>(jsonFixture);
    var kafkaMessage = ConstructKafkaMessage(message);
    await _producer.ProduceAsync(publishTopicName, kafkaMessage);
}

// 容器启动、Topic创建、数据库初始化等其他方法省略

测试示例：

var someTopic = kafkaConfiguration.ProducerTopics[nameof(SomeClass)];
var someOtherTopic = kafkaConfiguration.ConsumerTopics[nameof(SomeClass2)];

// Act
_ = Task.Run(async () => await _host.RunAsync());
await ProduceTestMessageAsync<Class>(someTopic, data);

// Assert
var message = ConsumeTestMessage(someOtherTopic, 5000, 10)?.Message?.Value;

可能的原因与修复方案 #

1. Host未完全启动就发送消息 #

测试中用Task.Run启动Host后直接发送消息，此时后台Kafka消费者可能还未完成订阅，导致消息丢失。

修复：
等待Host启动完成后再发送消息，通过IHostApplicationLifetime监听启动完成事件：

var hostTask = Task.Run(async () => await _host.RunAsync());
var lifetime = _host.Services.GetRequiredService<IHostApplicationLifetime>();
var startedCompletionSource = new TaskCompletionSource();
lifetime.ApplicationStarted.Register(() => startedCompletionSource.SetResult());
await startedCompletionSource.Task;

// 再执行消息生产
await ProduceTestMessageAsync<Class>(someTopic, data);

2. 消费重试逻辑未等待延迟 #

ConsumeTestMessage中的Task.Delay未加await，导致循环无间隔重试，可能在消息同步完成前就终止。

修复：
添加await确保重试间隔生效：

counter++;
await Task.Delay(timeoutMilliseconds);

3. Topic创建后未等待元数据同步 #

Kafka创建Topic后需要时间同步元数据，立即生产消息可能触发隐性的"Topic不存在"错误。

修复：
创建Topic后添加等待，或主动验证Topic存在：

private async Task CreateTopicsAsync()
{
    try
    {
        using var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = _bootstrapServer }).Build();
        var topicSpecs = _topics.Select(topic => new TopicSpecification() { Name = topic, NumPartitions = 1, ReplicationFactor = 1 }).ToList();
        await adminClient.CreateTopicsAsync(topicSpecs);
        
        // 等待元数据同步
        await Task.Delay(1000);
        
        // 可选：主动验证Topic状态
        var metadata = await adminClient.GetMetadata(TimeSpan.FromSeconds(5));
        foreach (var topic in _topics)
        {
            while (!metadata.Topics.Any(t => t.Topic == topic))
            {
                await Task.Delay(200);
                metadata = await adminClient.GetMetadata(TimeSpan.FromSeconds(5));
            }
        }
    }
    catch (CreateTopicsException e)
    {
        // 仅忽略Topic已存在的错误，其他错误抛出
        if (!e.Results.All(r => r.Error.Code == ErrorCode.TopicAlreadyExists))
        {
            throw;
        }
    }
}

4. TearDown阶段删除所有Topic干扰后续测试 #

当前DeleteTopicsAsync删除Kafka中所有Topic，可能影响未完全清理的当前测试资源，或干扰下一个测试的初始化。

修复：
仅删除当前测试创建的Topic：

private async Task DeleteTopicsAsync()
{
    using var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = _bootstrapServer }).Build();
    await adminClient.DeleteTopicsAsync(_topics);
}

5. Kafka容器就绪判断不可靠 #

用UntilPortIsAvailable仅判断端口开放，不代表Kafka服务已完全启动（比如Kraft控制器选举完成）。

修复：
改用AdminClient验证服务就绪：

private async Task StartKafkaContainerAsync()
{
    var kafkaContainerName = $"kafka-integration_{_fixture.Create<string>()}";
    _kafkaContainer = new ContainerBuilder()
        // 原有容器配置保留
        .WithWaitStrategy(Wait.ForUnixContainer()
            .UntilCustomSucceed(async () =>
            {
                try
                {
                    using var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = _bootstrapServer }).Build();
                    var metadata = await adminClient.GetMetadata(TimeSpan.FromSeconds(1));
                    return metadata.Brokers.Any();
                }
                catch
                {
                    return false;
                }
            })
            .WithTimeout(TimeSpan.FromMinutes(2)))
        .Build();

    await _kafkaContainer.StartAsync();
}

6. 内存数据库初始化未处理异步逻辑 #

InitializeDbsAsync是同步方法，内部数据库初始化可能未完成就进入测试逻辑。

修复：
改为异步方法并等待初始化完成：

[SetUp]
public async Task SetUp()
{
    // ... 其他逻辑 ...
    await InitializeDbsAsync();
    // ... 其他逻辑 ...
}

private async Task InitializeDbsAsync()
{
    var dbName = $"InMemoryDb_{_fixture.Create<string>()}";
    _hostBuilder.ConfigureServices(delegate (IServiceCollection services)
    {
        _refContext = CreateInMemoryReferenceDataContext(dbName);
        InitInMemoryAppDataContext<TDataContext>(services, dbName);
    });
    
    var serviceProvider = _hostBuilder.BuildServiceProvider();
    _dataContext = serviceProvider.GetRequiredService<TDataContext>();
    await _dataContext.Database.EnsureCreatedAsync();
    await _refContext.Database.EnsureCreatedAsync();
}

总结 #

随机失败的核心原因大多是异步操作未正确等待、资源未完全就绪或隔离不彻底。优先检查Host启动、消费重试、Topic同步这几个环节，这些是集成测试中最容易出现时序问题的点。

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