0.7.0 · Stable

Kafka and Schema Registry

Your application works with generated C# models. The Kafka Avro serializer handles schema registration and the message format. Configure it for the same runtime you selected for generation, then produce and consume your model as usual.

Before you start

Use the User schema from Getting started, a development Kafka broker, and Schema Registry. Create a dedicated users topic for this example. The defaults are localhost:9092 and http://localhost:8081; set KAFKA_BOOTSTRAP_SERVERS and SCHEMA_REGISTRY_URL to use other endpoints.

If you have the Kafka CLI installed, create the topic with:

kafka-topics --bootstrap-server localhost:9092 --create --topic users --partitions 1 --replication-factor 1

Choose one of the following integrations.

Apache.Avro

Keep Apache.Avro selected for generation and install Confluent’s Avro serializer:

dotnet add package Confluent.SchemaRegistry.Serdes.Avro --version 2.15.1

The generated User implements ISpecificRecord, which is the contract this serializer expects. The following round-trip example configures a typed producer and consumer, sends a User, and verifies that the matching message comes back with the same values:

using Confluent.Kafka;
using Confluent.Kafka.SyncOverAsync;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;
using example;

var brokers = Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP_SERVERS") ?? "localhost:9092";
var registryUrl = Environment.GetEnvironmentVariable("SCHEMA_REGISTRY_URL") ?? "http://localhost:8081";
const string topic = "users";

using var registry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { Url = registryUrl });
using var producer = new ProducerBuilder<string, User>(new ProducerConfig
    {
        BootstrapServers = brokers,
        MessageTimeoutMs = 15000,
    })
    .SetKeySerializer(Serializers.Utf8)
    .SetValueSerializer(new AvroSerializer<User>(registry).AsSyncOverAsync())
    .Build();

using var consumer = new ConsumerBuilder<string, User>(new ConsumerConfig
    {
        BootstrapServers = brokers,
        GroupId = $"user-demo-{Guid.NewGuid():N}",
        AutoOffsetReset = AutoOffsetReset.Earliest,
        EnableAutoCommit = false,
    })
    .SetKeyDeserializer(Deserializers.Utf8)
    .SetValueDeserializer(new AvroDeserializer<User>(registry).AsSyncOverAsync())
    .Build();

using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(45));
var key = Guid.NewGuid().ToString("N");
var original = new User { Name = "Ada", Email = "ada@example.com" };

await producer.ProduceAsync(topic, new Message<string, User> { Key = key, Value = original }, timeout.Token);
consumer.Subscribe(topic);

User restored;
while (true)
{
    var received = consumer.Consume(timeout.Token);
    if (received.Message.Key != key) continue;
    restored = received.Message.Value;
    break;
}

consumer.Close();
if (restored.Name != original.Name || restored.Email != original.Email)
    throw new InvalidOperationException("The consumed User differs from the produced User.");

Console.WriteLine($"Round trip: {restored.Name} <{restored.Email}>");

Chr.Avro

Use this integration when your application uses Chr.Avro. Remove Apache.Avro if it was only installed for the quick start, install the Chr Kafka integration, and select the generator target:

dotnet remove package Apache.Avro
dotnet add package Chr.Avro.Confluent --version 10.13.1

<PropertyGroup>
  <AvroSourceGeneratorAvroLibrary>Chr</AvroSourceGeneratorAvroLibrary>
</PropertyGroup>

The Chr producer registers its value schema under users-value. The consumer resolves the writer schema through the same registry:

using Chr.Avro.Confluent;
using Confluent.Kafka;
using Confluent.SchemaRegistry;
using example;

var brokers = Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP_SERVERS") ?? "localhost:9092";
var registryUrl = Environment.GetEnvironmentVariable("SCHEMA_REGISTRY_URL") ?? "http://localhost:8081";
const string topic = "users";

using var registry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { Url = registryUrl });
var producerBuilder = new ProducerBuilder<string, User>(new ProducerConfig
    {
        BootstrapServers = brokers,
        MessageTimeoutMs = 15000,
    })
    .SetKeySerializer(Serializers.Utf8);

producerBuilder = await producerBuilder.SetAvroValueSerializer(
    registry, $"{topic}-value", AutomaticRegistrationBehavior.Always);
using var producer = producerBuilder.Build();

using var consumer = new ConsumerBuilder<string, User>(new ConsumerConfig
    {
        BootstrapServers = brokers,
        GroupId = $"user-demo-{Guid.NewGuid():N}",
        AutoOffsetReset = AutoOffsetReset.Earliest,
        EnableAutoCommit = false,
    })
    .SetKeyDeserializer(Deserializers.Utf8)
    .SetAvroValueDeserializer(registry)
    .Build();

using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(45));
var key = Guid.NewGuid().ToString("N");
var original = new User { Name = "Ada", Email = "ada@example.com" };

await producer.ProduceAsync(topic, new Message<string, User> { Key = key, Value = original }, timeout.Token);
consumer.Subscribe(topic);

User restored;
while (true)
{
    var received = consumer.Consume(timeout.Token);
    if (received.Message.Key != key) continue;
    restored = received.Message.Value;
    break;
}

consumer.Close();
if (restored.Name != original.Name || restored.Email != original.Email)
    throw new InvalidOperationException("The consumed User differs from the produced User.");

Console.WriteLine($"Round trip: {restored.Name} <{restored.Email}>");

Both examples print:

Round trip: Ada <ada@example.com>

The unique message key lets the consumer find the message sent by this run. A 45-second timeout prevents the example from waiting indefinitely. Each run uses a fresh consumer group and reads from the start of the demo topic; use your application’s normal group and offset policy in production.

Schemas and registration

These development examples allow automatic schema registration. If schemas are provisioned separately in your environment, configure the serializer’s registration behavior accordingly. Producer and consumer must agree on the Avro schema contract, and the runtime must support your logical types and union branches.

For Chr unions represented by generated variant interfaces, register the generated serializer and deserializer builder cases before the default Chr cases. See the version’s union examples and generated union types.

Complete examples with local containers

The repository includes complete Kafka and Schema Registry examples, including Testcontainers setup:

• Apache.Avro Kafka sample.
• Chr.Avro Kafka sample.

These links match the selected package version. Repository samples reference the generator project and use the SDK selected by their checkout’s global.json; their build requirements can differ from an application consuming the NuGet package. Docker with Linux containers is required for the container examples.