如何确保ElasticSearch集成测试的确定性？解决索引验证不稳定问题

问题背景 #

在对基于Elasticsearch的数据存储类做集成测试时，出现测试结果不稳定的情况：多数时候测试能通过（数据确实已写入ES），但偶尔会失败，重新运行又恢复正常。测试流程为通过被测类写入数据后，验证ES中的数据是否符合预期，示例代码如下：

测试代码示例 #

[SetUp]
public void Setup()
{
    ElasticClient.DeleteByQuery(Indices.Index(IndexName), d =>
        d.Query(x => x.MatchAll())
    );
}

[Test]
public async Task GetAllConversations_ShouldReturnAllConversations()
{
    // Arrange
    Guid[] randomGuids = new[]
    {
        Guid.NewGuid(),
        Guid.NewGuid(),
        Guid.NewGuid(),
        Guid.NewGuid(),
    };

    foreach (var randomGuid in randomGuids)
    {
        await _conversationStorage.AddMessage(randomGuid, new Message()
        {
            Text = "test",
            Type = MessageType.UserMade
        });
    }

    // Act
    var result = await _sut.GetAllConversations();
    
    // Assert
    Assert.That(result?.Conversations.Count, Is.EqualTo(4));
}

（AddMessage内部调用IndexAsync）

使用的技术栈：Elastic.Clients.Elasticsearch .NET库、Elasticsearch 8.10.2版本。需要确保所有写入数据完成索引后，再执行断言检查。

解决方案 #

1. 写入时指定等待刷新策略 #

在IndexAsync调用中设置Refresh.WaitFor，让写入操作等待ES完成索引刷新后再返回，确保数据立即可见：

// 修改AddMessage内部的IndexAsync调用
await _elasticClient.IndexAsync(document, idx => idx
    .Index(IndexName)
    .Refresh(Refresh.WaitFor)
);

该参数会阻塞直到索引刷新完成，彻底避免写入后查询时数据未被索引的问题。

2. 写入后手动触发刷新 #

如果不想修改业务代码的写入逻辑，可在测试的Arrange阶段完成数据写入后，手动触发索引刷新并等待集群状态同步：

// 数据写入后添加以下代码
await _elasticClient.Indices.RefreshAsync(IndexName);
// 可选：等待集群状态变为Green，确保分片完全同步
await _elasticClient.Cluster.HealthAsync(h => h
    .WaitForStatus(WaitForStatus.Green)
    .Index(IndexName)
);

3. 重试查询直到满足预期 #

若对测试性能有要求，可在断言前添加重试逻辑，直到查询结果符合条件或超时：

// Act阶段替换为重试逻辑
var result = await RetryUntilSuccessAsync(async () =>
{
    var res = await _sut.GetAllConversations();
    return res?.Conversations.Count == 4 ? res : null;
}, TimeSpan.FromSeconds(5));

// Assert
Assert.That(result?.Conversations.Count, Is.EqualTo(4));

// 重试方法实现
private async Task<T> RetryUntilSuccessAsync<T>(Func<Task<T>> action, TimeSpan timeout)
{
    var startTime = DateTime.UtcNow;
    while (DateTime.UtcNow - startTime < timeout)
    {
        var result = await action();
        if (result != null)
            return result;
        await Task.Delay(100);
    }
    throw new TimeoutException("查询超时，未获取到预期结果");
}

这种方式通过短间隔重试，适配ES的异步索引延迟，同时避免强制刷新带来的性能损耗。

4. 调整测试环境索引设置 #

在测试环境中，先禁用索引自动刷新，测试时手动触发刷新，减少自动刷新的开销：

// 测试初始化时设置
await _elasticClient.Indices.UpdateSettingsAsync(IndexName, s => s
    .Settings(st => st
        .IndexRefreshInterval("-1")
    )
);

// 数据写入后手动刷新
await _elasticClient.Indices.RefreshAsync(IndexName);

// 测试清理时恢复默认设置（可选）
await _elasticClient.Indices.UpdateSettingsAsync(IndexName, s => s
    .Settings(st => st
        .IndexRefreshInterval("1s")
    )
);

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