Batch Production

When you need to send multiple messages, Dekaf provides efficient batch APIs that handle the complexities of concurrent production.

The Challenge with ValueTask

ProduceAsync returns a ValueTask<RecordMetadata> for performance reasons - it can complete synchronously when messages are added to an internal buffer. However, this creates a challenge when sending multiple messages:

// ❌ Avoid this - converting every ValueTask allocates a Task when it does not complete synchronously
var tasks = messages.Select(m => producer.ProduceAsync("topic", m.Key, m.Value).AsTask());
await Task.WhenAll(tasks);

ProduceAllAsync

Dekaf provides ProduceAllAsync which handles the ValueTask conversion internally:

// ✅ Use ProduceAllAsync for batch sends
var results = await producer.ProduceAllAsync(messages);

With Message Objects

var messages = new[]
{
    new ProducerMessage<string, string> { Topic = "orders", Key = "order-1", Value = orderJson1 },
    new ProducerMessage<string, string> { Topic = "orders", Key = "order-2", Value = orderJson2 },
    new ProducerMessage<string, string> { Topic = "orders", Key = "order-3", Value = orderJson3 },
};

var results = await producer.ProduceAllAsync(messages);

// Results are in the same order as input
for (int i = 0; i < results.Length; i++)
{
    Console.WriteLine($"Message {i}: partition {results[i].Partition}, offset {results[i].Offset}");
}

With Tuples (Same Topic)

When all messages go to the same topic, use the tuple overload to avoid creating message objects:

var keyValuePairs = new[]
{
    ("order-1", orderJson1),
    ("order-2", orderJson2),
    ("order-3", orderJson3),
};

var results = await producer.ProduceAllAsync("orders", keyValuePairs);

From a Collection

Works with any IEnumerable:

var orders = await GetPendingOrdersAsync();

var messages = orders.Select(order =>
    new ProducerMessage<string, string>
    {
        Topic = "orders",
        Key = order.Id,
        Value = JsonSerializer.Serialize(order)
    }
);

var results = await producer.ProduceAllAsync(messages);

Error Handling

If any message fails, the entire ProduceAllAsync throws an AggregateException:

try
{
    var results = await producer.ProduceAllAsync(messages);
}
catch (AggregateException ex)
{
    foreach (var inner in ex.InnerExceptions)
    {
        if (inner is ProduceException produceEx)
        {
            Console.WriteLine($"Failed to produce: {produceEx.Message}");
        }
    }
}

Partial Success Handling

If you need to handle partial failures (some messages succeed, some fail), use explicit Task conversion:

var tasks = messages.Select(async m =>
{
    try
    {
        return (m, Result: await producer.ProduceAsync(m).AsTask(), Error: (Exception?)null);
    }
    catch (Exception ex)
    {
        return (m, Result: default(RecordMetadata), Error: ex);
    }
}).ToList();

var results = await Task.WhenAll(tasks);

var succeeded = results.Where(r => r.Error == null).ToList();
var failed = results.Where(r => r.Error != null).ToList();

Console.WriteLine($"Succeeded: {succeeded.Count}, Failed: {failed.Count}");

Performance Considerations

Batch Size

Sending many messages at once is more efficient than sending them one by one:

// ❌ Inefficient - waits for each message
foreach (var msg in messages)
{
    await producer.ProduceAsync("topic", msg.Key, msg.Value);
}

// ✅ Efficient - sends all at once
await producer.ProduceAllAsync(messages);

Memory Usage

ProduceAllAsync materializes the input enumerable to get a count. For very large collections, consider batching:

var allMessages = GetMillionsOfMessages();

// Process in batches to control memory
foreach (var batch in allMessages.Chunk(1000))
{
    await producer.ProduceAllAsync(batch);
}

Combining with Fire-and-Forget

For maximum throughput when you don't need results, combine FireAsync() with FlushAsync(). Awaiting FireAsync() waits only for local enqueue and backpressure, not broker acknowledgment:

// Queue all messages without waiting for broker acknowledgment
foreach (var msg in messages)
{
    await producer.FireAsync("topic", msg.Key, msg.Value);
}

// Ensure all are delivered before continuing
await producer.FlushAsync();

Real-World Example

Here's a complete example processing a batch of orders:

public sealed class OrderBatchProcessor(
    IKafkaProducer<string, Order> producer,
    ILogger<OrderBatchProcessor> logger)
{
    public async Task ProcessOrderBatchAsync(IReadOnlyList<Order> orders)
    {
        var messages = orders.Select(order => new ProducerMessage<string, Order>
        {
            Topic = "orders",
            Key = order.Id,
            Value = order,
            Headers = Headers.Create()
                .Add("source", "batch-processor")
                .Add("batch-size", orders.Count.ToString())
        }).ToList();

        try
        {
            var results = await producer.ProduceAllAsync(messages);

            logger.LogInformation(
                "Sent {Count} orders across {Partitions} partitions",
                results.Length,
                results.Select(r => r.Partition).Distinct().Count());
        }
        catch (AggregateException ex)
        {
            logger.LogError(ex, "Failed to send some orders in batch");
            throw;
        }
    }
}