[Bug Report]: BatchProduceAsync seems to produce items out or order

[AlexeyRaga]

Prerequisites

• [X] I have searched issues to ensure it has not already been reported

Description

It looks like BatchProduceAsync can produce message out-of-order. It happens from time to time, and the order doesn't seem to be deterministic. Most of the time it does seem to produce messages in the right order, but not always.

Steps to reproduce

I was using this code to reproduce:

private sealed record Batch(Guid Id, List<ISpecificRecord> Messages);

private static Batch CreateBatch()
{
    var id = Guid.NewGuid();
    var messages = new List<ISpecificRecord>
    {
        new UserJoined { id = id, order = 1 },
        new UserDeactivated { id = id, order = 2 },
        new UserActivated { id = id, order = 3 }
    };

    return new(id, messages);
}

[Fact]
public async Task Should_preserve_order_of_messages()
{
    var batchesToSend = Enumerable.Range(0, 10).Select(_ => CreateBatch()).ToList();

    foreach (var batch in batchesToSend)
    {
        var toProduce =
            batch.Messages
                .Select(x => new BatchProduceItem(
                    fixture.TopicName,
                    batch.Id.ToString(),
                    x,
                    new MessageHeaders()))
                .ToList();

        await fixture.Producer.BatchProduceAsync(toProduce);
    }

    // assertion is omitted for brevity 
}

I then look at the topic itself and see that some batches come out of order.

Expected behavior

I expect that the messages order is guaranteed by the batch producer all the time.

Actual behavior

Most often messages are coming in the right order: 1, 2, 3 but sometimes they are written in the wrong order.

KafkaFlow version

v3.0.10