Eventso.KafkaProducer 2.5.2

Binary low-allocating Kafka producer for .Net

Binary Kafka producer with a unified api that accepts ReadOnlySpan<byte> instead of Message<TKey,TValue> and avoids creating intermediate byte arrays during serialization. Uses low-level api of Confluent.Kafka library and is fully compatible with it. A single producer instance is enough for your entire application.

Features

• Binary producer api accepts ReadOnlySpan<byte> for key and value
• Only single producer instance in application is enough
• Effectively producing message batch
• Ready to use overloads for common key types: short, int, long, string, Guid
• Support for Protobuf, System.Text.Json, SpanJson serialization (separate packages)
• Confluent.Kafka compatibility
• Tested in heavy-load production environment

Registration

There are 3 ways to create binary producer:

// 1. Create from Confluent.Kafka ProducerBuilder<TAnyKey, TAnyValue>

ProducerBuilder<byte[], byte[]> confluentBuilder = ...

IProducer binaryProducer = confluentBuilder.BuildBinary();

// 2. Using non-generic ProducerBuilder with same api as ProducerBuilder<TKey, TValue>

ProducerBuilder producerBuilder = new ProducerBuilder(producerConfig);

producerBuilder
    .SetDefaultPartitioner(...)
    .SetErrorHandler(...);

IProducer binaryProducer = producerBuilder.Build();

// 3. Create binary producer from generic Confluent.Kafka producer

IProducer<TAnyKey, TAnyValue> confluentProducer = ...

IProducer binaryProducer = confluentProducer.CreateBinary();

How to use the producer:

There are two basic methods in IProducer interface

Task<DeliveryResult> ProduceAsync(
    string topic,
    ReadOnlySpan<byte> key,
    ReadOnlySpan<byte> value,
    CancellationToken cancellationToken = default(CancellationToken),
    Headers? headers = null,
    Timestamp timestamp = default,
    Partition? partition = null);

void Produce(
    string topic,
    ReadOnlySpan<byte> key,
    ReadOnlySpan<byte> value,
    Headers? headers = null,
    Timestamp timestamp = default,
    Action<DeliveryReport>? deliveryHandler = null,
    Partition? partition = null);

Library contains extension methods for frequently used key types: short, int, long, string, Guid

MessageBatch

MessageBatch creates only one TaskCompletionSource and Task per batch while original producer api creates TaskCompletionSource per message. Batch Produce supports all producer features.

var batch = producer.CreateBatch(topicName);

foreach(var record in records)
{
    batch.Produce(record.Key, record.Value)
}

//wait for delivery all messages
await batch.Complete(token);

Protobuf and Json values

Additional packages contain method overloads that accepts Google.Protobuf.IMessage or typed object as message value. They use stack or ArrayPool for non-allocating serialization.

• Protobuf
• System.Text.Json (methods also accepts optional JsonSerializerOptions or JsonSerializerContext)
• SpanJson

Performance and memory allocation benchmark

Benchmark source code

Thanks to @xeromorph for the great ideas and help.