0.8.0 · Preview

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

Use the Apache.Avro integration below.

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}>");

The example prints:

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

This development example allows 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.

Complete examples with local containers

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

This link matches 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.