Introduction
As a .NET developer, when you first start using Kafka, the examples you usually find are simple- just sending and receiving basic messages. But in real-world applications, the use cases, requirements, and architecture are usually more complex or get more complex with time. You’ll need to ensure the data in Kafka topics is well formed and valid, handle errors more properly with retires and fallbacks, and sometimes deal with topics containing different messages.
In this post, I’ll cover these advanced topics. We’ll look at how to use Confluent’s Schema Registry, handle multiple message types in a single topic, and build proper error handling with retries and dead letter queues (DLQ).
If you’re new to Kafka in .NET, I recommend checking out my previous blog post, in which I explain the basics of Kafka, how to create a producer and consumer, and how to get a simple setup running. You can treat that post as a starting point before continuing here.
You don’t need to be a Kafka expert to follow along with this post, but you should already be comfortable writing basic producers and consumers in .NET. You’ll also need a running Kafka and Schema Registry instance (for example, using Docker or Confluent Cloud).
What is Schema Registry and Why It Matters?
When you send data to Kafka, it’s important to understand that Kafka doesn’t care about what the data is or how it’s structured. At its core, Kafka works only with bytes. It doesn’t know if the data you’re sending is a string, an integer, or a more complex type.
Because of this—and to keep producers and consumers decoupled—they don’t talk to each other directly. Instead, they communicate through topics. But the consumer still needs to know what kind of data was sent in order to deserialize it correctly.
This can become a problem. Imagine the producer changes the structure of the data or starts sending invalid data. In that case, the consumer might fail because it no longer understands how to process the message.
This is where Schema Registry and schemas come in. Schema Registry is a separate service that keeps track of the data structure (the schema) that both producers and consumers agree on.
When a producer wants to send a message, it first checks with the Schema Registry to see if the schema already exists. If not, it registers the schema. Then it serializes the data using that schema and sends it to Kafka, along with a schema ID.
When a consumer reads the message, it uses the schema ID to fetch the correct schema from the Schema Registry and deserializes the data accordingly. If the data doesn’t match the expected schema, Schema Registry will raise an error.
Supported Data Formats
Schema Registry supports several data formats, with the most common being Avro, JSON Schema, and Protobuf. Each format has its own trade-offs.
- Avro is compact and efficient, making it popular for high-throughput systems.
- JSON Schema is human-readable and easier to work with when debugging or during development.
- Protobuf is more strict and efficient, often used in strongly typed systems.
Your choice depends on what matters most in your system—performance, flexibility, or ease of use. In .NET, all of these formats are supported through Confluent’s serializers and deserializer.
Setting up Kafka and Schema Registry
Before we dive into code, let’s setup Kafka and Schema Registry. We’ll use Docker Compose to spin up the required service locally. This is the easiest way to get started without installing anything manually.
You’ll need three services:
- Zookeeper – required by Kafka (unless you’re using KRaft)
- Kafka – the actual message broker
- Schema Registry – manages Avro schemas for producers and consumers
Use the following docker-compose.yml file:
version: '3.8'
services:
zookeeper:
image: confluentinc/cp-zookeeper:7.5.1
hostname: zookeeper
container_name: zookeeper
ports:
- "2181:2181"
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
kafka:
image: confluentinc/cp-kafka:7.5.1
hostname: kafka
container_name: kafka
depends_on:
- zookeeper
ports:
- "9092:9092" # For your .NET app running on the host
- "29092:29092" # For Schema Registry inside Docker
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
schema-registry:
image: confluentinc/cp-schema-registry:7.5.1
hostname: schema-registry
container_name: schema-registry
depends_on:
- kafka
ports:
- "8081:8081"
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:29092
Why Dual Listeners?
You might be wondering why Kafka is exposing two different ports and listeners.
This setup solves a common issue when working with containers and local development tools at the same time:
PLAINTEXT://kafka:29092is used by other containers, like Schema Registry. Inside Docker, services can reach each other by container name (kafka,schema-registry, etc.).PLAINTEXT_HOST://localhost:9092is used by your .NET app running on your machine (not inside Docker). It connects to Kafka through your host’s network.
Without this setup, either your Schema Registry will fail to resolve localhost, or your .NET app will fail to resolve kafka. So this dual-listener configuration keeps both environments happy.
Using Confluent Schema Registry with .NET
Now that we have Kafka and Schema Registry running, let’s look at how to integrate them in a .NET app using Avro.
We’ll go step by step:
- Set up Schema Registry in code
- Produce a message using Avro serialization
- Consume it using Avro deserialization
We’re using the official Confluent .NET client and Schema Registry serializers.
Setting up Schema Registry in .NET
First, install the required NuGet packages:
dotnet add package Confluent.Kafka
dotnet add package Confluent.SchemaRegistry
dotnet add package Confluent.SchemaRegistry.Serdes.Avro
These give you:
- Kafka producer/consumer APIs
- Schema Registry client
- Avro serializers/deserializers
Then, create a configuration section (e.g., KafkaSettings) with:
{
"KafkaSettings": {
"BootstrapServers": "localhost:9092",
"SchemaRegistryUrl": "http://localhost:8081",
"GroupId": "user-consumer-group"
}
}
Serializing and Deserializing Messages using Avro
Next, we’ll create the message class that will be shared between the producer and consumer. Since we’re using Schema Registry, we need to make sure the class is compatible with one of the supported serialization formats. In this post, we’ll use Avro, which means the class must implement ISpecificRecord.
The easiest way to do that is by generating the class from an Avro schema (.avsc) file using avrogen.
You can install avrogen globally using the following command:
dotnet tool install --global Apache.Avro.Tools --version 1.12.0
Then, create an Avro schema file (with the .avsc extension). For example:
{
"namespace": "KafkaSchemas",
"type": "record",
"name": "UserCreated",
"fields": [
{
"name": "UserId",
"type": "string"
},
{
"name": "UserName",
"type": "string"
},
{
"name": "Email",
"type": "string"
}
]
}
Next, use the command line to generate the C# class;
avrogen -s .\UserCreated.avsc .
After generating the C# class, you can use it like any other DTO.
Creating the producer
Now we can continue and create the producer. The producer will use Schema Registry to manage and validate the message schema. When the producer sends a message, the serializer will first check if the schema already exists in the registry. If it doesn’t, it will register the schema automatically (if allowed), and then serialize the message using Avro. Here’s how to configure and use the producer in .NET:
using Confluent.Kafka;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;
using KafkaSchemas;
using Microsoft.Extensions.Configuration;
// Load configuration from appsettings.json
var config = new ConfigurationBuilder()
.AddJsonFile("appsettings.json", optional: true)
.Build();
// Bind the KafkaSettings section to a strongly typed class
var kafkaSettings = config.GetSection("KafkaSettings").Get<KafkaSettings>();
// Configure Schema Registry
var schemaRegistryConfig = new SchemaRegistryConfig
{
Url = kafkaSettings.SchemaRegistryUrl
};
// Configure Kafka producer settings
var producerConfig = new ProducerConfig
{
BootstrapServers = kafkaSettings.BootstrapServers,
ClientId = "user-producer", // You can use kafkaSettings.ClientId if exposed
EnableIdempotence = true, // Ensures no duplicate messages
Acks = Acks.All, // Wait for all replicas to acknowledge
CompressionType = CompressionType.Gzip // Compress messages to reduce size
};
// Create a schema registry client
using var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
// Build the Kafka producer and attach the Avro serializer for message values
using var producer = new ProducerBuilder<string, UserCreated>(producerConfig)
.SetValueSerializer(new AvroSerializer<UserCreated>(schemaRegistry))
.Build();
bool _running = true;
// Handle CTRL+C to allow graceful shutdown
Console.CancelKeyPress += (_, args) =>
{
_running = false;
args.Cancel = true;
Console.WriteLine("\nCTRL+C detected. Exiting...");
};
// Continuously send a test user message every 2 seconds
while (_running)
{
Console.WriteLine("Sending user...");
var message = new UserCreated
{
UserId = Guid.NewGuid().ToString(),
UserName = "John",
Email = "JohnDoe@example.com"
};
// Produce the message to the configured Kafka topic
await producer.ProduceAsync(kafkaSettings.TopicName, new Message<string, UserCreated>
{
Key = message.UserId,
Value = message
});
await Task.Delay(2000); // Wait for 2 seconds before sending the next message
}
This code sets up a Kafka producer in .NET that sends Avro-serialized UserCreated messages to a Kafka topic. It uses Confluent’s Schema Registry to manage and version the message schema. Before sending, the message is serialized using the Avro format and registered with the Schema Registry, allowing consumers to deserialize it correctly based on the schema ID embedded in the message.
Creating the consumer
In our example, the consumer will be a simple background worker service that subscribes to the Kafka topic and uses Schema Registry to deserialize messages using the correct Avro schema. The consumer will read messages from the topic, deserialize them into UserCreated objects, and log the output to the console.
To make this work, we’ll configure the consumer to use the same Schema Registry as the producer, and set up an AvroDeserializer<UserCreated> to correctly interpret the schema attached to each message.
This is the consumer class code. Notice how we use Schema Registry with the consumer:
using Confluent.Kafka;
using Confluent.Kafka.SyncOverAsync;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;
using KafkaSchemas; // Namespace for the generated Avro UserCreated class
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace ConsumerWorker
{
public class Worker : BackgroundService
{
private readonly ILogger<Worker> _logger;
private readonly KafkaSettings _kafkaSettings;
private IConsumer<string, UserCreated>? _consumer;
private ISchemaRegistryClient? _schemaRegistry;
private readonly string _topic;
public Worker(ILogger<Worker> logger, IOptions<KafkaSettings> kafkaSettings)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_kafkaSettings = kafkaSettings.Value ?? throw new ArgumentNullException(nameof(kafkaSettings));
_topic = _kafkaSettings.TopicName ?? "user-events";
}
public override Task StartAsync(CancellationToken cancellationToken)
{
// Set up the Kafka consumer configuration
var consumerConfig = new ConsumerConfig
{
BootstrapServers = _kafkaSettings.BootstrapServers,
GroupId = _kafkaSettings.GroupId,
AutoOffsetReset = AutoOffsetReset.Earliest,
EnableAutoCommit = true
};
// Configure access to Schema Registry
var schemaRegistryConfig = new SchemaRegistryConfig
{
Url = _kafkaSettings.SchemaRegistryUrl
};
// Create a cached Schema Registry client
_schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
// Create the consumer with Avro deserializer (sync wrapper over async API)
_consumer = new ConsumerBuilder<string, UserCreated>(consumerConfig)
.SetValueDeserializer(new AvroDeserializer<UserCreated>(_schemaRegistry).AsSyncOverAsync())
.Build();
// Subscribe to the target topic
_consumer.Subscribe(_topic);
_logger.LogInformation("Kafka consumer started and subscribed to topic: {Topic}", _topic);
return base.StartAsync(cancellationToken);
}
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
return Task.Run(() =>
{
try
{
while (!stoppingToken.IsCancellationRequested)
{
try
{
// Attempt to consume a message
var result = _consumer?.Consume(stoppingToken);
if (result != null)
{
var user = result.Message.Value;
// Log the received user data
_logger.LogInformation("Received user: {UserId} - {Email}", user.UserId, user.Email);
}
}
catch (ConsumeException ex)
{
// Handle known Kafka consume errors
_logger.LogError(ex, "Error while consuming message.");
}
}
}
catch (OperationCanceledException)
{
// Triggered by cancellation token on shutdown
_logger.LogInformation("Kafka consumer stopping...");
}
finally
{
// Graceful shutdown and cleanup
_consumer?.Close();
_consumer?.Dispose();
_schemaRegistry?.Dispose();
}
}, stoppingToken);
}
}
}
This background worker acts as a Kafka consumer that listens to the user-events topic. It connects to the Kafka broker and the Confluent Schema Registry, then continuously reads Avro-encoded messages.
The key part is the use of AvroDeserializer<UserCreated>, which communicates with Schema Registry using the schema ID embedded in each message. When a message is received, the consumer fetches the correct schema from Schema Registry (if not already cached) and deserializes the binary payload into a strongly typed UserCreated object. This ensures that both producer and consumer are aligned on the schema and can evolve safely over time.
Effective Error Handling in Kafka with .NET
When working with Kafka in production (and any message broker or software), things don’t always go as expected. Errors and exceptions will probably happen and you need to handle network failures and issues, serialization and deserialization failures or schema mismatches that could cause your consumers to crash. To build a reliable Kafka-based system, you need a strategy for handling these errors gracefully.
In this section, we’ll cover:
- Common Kafka errors and what causes them
- How to implement retries and backoff
- How to use a Dead Letter Queue (DLQ) to isolate bad messages
- A code example with proper error handling and fallback
Common Kafka Errors and Their Causes
Here are a few types of errors you’re likely to run into:
- Deserialization Errors – Occur when the schema in the message doesn’t match the expected one, or when the data is malformed.
- Network Errors – Temporary issues when connecting to brokers or the schema registry.
- Offset Commit Failures – Happen when a consumer tries to commit an offset after a failure, or if auto-commit is misused.
- Application Exceptions – Your own business logic may fail while processing a valid message.
Handling Retries and Backoff Strategies
Instead of crashing on the first failure, it’s better to retry transient errors a few times. A common approach is to use exponential backoff, where you increase the delay between retries to give the system time to recover.
You can implement retries manually inside your consumer loop or use a resilience library like Polly.
First, we need to add the polly policy definition:
private readonly AsyncRetryPolicy _retryPolicy;Then initialize it in our consumer:
// Retry policy for message processing failures, not for Kafka internal consumption
_retryPolicy = Policy
.Handle<Exception>() // Catch any processing exception (not Kafka Consume errors)
.WaitAndRetryAsync(3, retryAttempt =>
TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
(exception, timeSpan, retryCount, _) =>
{
_logger.LogWarning(exception,
"Processing failed. Retry {RetryCount} in {Delay}s.",
retryCount, timeSpan.TotalSeconds);
});
Now, we will wrap our message processing logic with Polly.
await _retryPolicy.ExecuteAsync(async () =>
{
// Simulate processing logic that could fail
await ProcessUserAsync(user);
});
Note – Why we did not wrap consume with polly?
If you do, you’re retrying the Consume() operation, which is not retryable in the way Polly is used. Consume just polls Kafka; it’s not a transient failure that benefits from retry.
The final consumer class now looks like this:
public class Worker : BackgroundService
{
private readonly ILogger<Worker> _logger;
private readonly KafkaSettings _kafkaSettings;
private readonly AsyncRetryPolicy _retryPolicy;
private IConsumer<string, UserCreated>? _consumer;
private ISchemaRegistryClient? _schemaRegistry;
private readonly string _topic;
public Worker(ILogger<Worker> logger, IOptions<KafkaSettings> kafkaSettings)
{
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_kafkaSettings = kafkaSettings.Value ?? throw new ArgumentNullException(nameof(kafkaSettings));
_topic = _kafkaSettings.TopicName ?? "user-events";
// Retry policy for message processing failures, not for Kafka internal consumption
_retryPolicy = Policy
.Handle<Exception>() // Catch any processing exception (not Kafka Consume errors)
.WaitAndRetryAsync(3, retryAttempt =>
TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
(exception, timeSpan, retryCount, _) =>
{
_logger.LogWarning(exception,
"Processing failed. Retry {RetryCount} in {Delay}s.",
retryCount, timeSpan.TotalSeconds);
});
}
public override Task StartAsync(CancellationToken cancellationToken)
{
var consumerConfig = new ConsumerConfig
{
BootstrapServers = _kafkaSettings.BootstrapServers,
GroupId = _kafkaSettings.GroupId,
AutoOffsetReset = AutoOffsetReset.Earliest,
EnableAutoCommit = true
};
var schemaRegistryConfig = new SchemaRegistryConfig
{
Url = _kafkaSettings.SchemaRegistryUrl
};
_schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
_consumer = new ConsumerBuilder<string, UserCreated>(consumerConfig)
.SetValueDeserializer(new AvroDeserializer<UserCreated>(_schemaRegistry).AsSyncOverAsync())
.Build();
_consumer.Subscribe(_topic);
_logger.LogInformation("Kafka consumer started and subscribed to topic: {Topic}", _topic);
return base.StartAsync(cancellationToken);
}
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
return Task.Run(async () =>
{
try
{
while (!stoppingToken.IsCancellationRequested)
{
try
{
var result = _consumer?.Consume(stoppingToken);
if (result == null) continue;
var user = result.Message.Value;
// Wrap only the message processing in Polly
await _retryPolicy.ExecuteAsync(async () =>
{
// Simulate processing logic that could fail
await ProcessUserAsync(user);
});
_logger.LogInformation("Successfully processed user: {UserId} - {Email}", user.UserId, user.Email);
}
catch (ConsumeException ex)
{
_logger.LogError(ex, "Kafka consume error.");
}
}
}
catch (OperationCanceledException)
{
_logger.LogInformation("Kafka consumer stopping...");
}
finally
{
_consumer?.Close();
_consumer?.Dispose();
_schemaRegistry?.Dispose();
}
}, stoppingToken);
}
private async Task ProcessUserAsync(UserCreated user)
{
// Add business logic here, can throw to trigger retry
await Task.Delay(10); // Simulate work
if (user.Email == "fail@example.com") throw new Exception("Simulated failure");
}
}
Implementing a Dead Letter Queue (DLQ)
When a message keeps failing after all retries, it’s better to move it to a special Kafka topic called a Dead Letter Queue (DLQ). This way, you don’t lose data, and your main consumer can continue processing healthy messages.
To be able to publish a DLQ, you need to create and initialize producer:
// Create a separate producer for sending failed messages to the DLQ
var producerConfig = new ProducerConfig
{
BootstrapServers = _kafkaSettings.BootstrapServers
};
_dlqProducer = new ProducerBuilder<string, string>(producerConfig).Build();
Then just use in the exception block :
try
{
await _retryPolicy.ExecuteAsync(async () =>
{
await ProcessUserAsync(user);
});
_logger.LogInformation("Successfully processed user: {UserId} - {Email}", user.UserId, user.Email);
}
catch (Exception ex)
{
_logger.LogError(ex, "Message failed after retries. Sending to DLQ...");
var dlqPayload = JsonSerializer.Serialize(user);
await _dlqProducer.ProduceAsync(_kafkaSettings.DlqTopicName, new Message<string, string>
{
Key = result.Message.Key,
Value = dlqPayload
});
_logger.LogInformation("Message sent to DLQ: {DlqTopic}", _kafkaSettings.DlqTopicName);
}
Managing Multi-Message Topics in .NET
Kafka topics often contain more than one type of message. This is common in systems where events related to the same business domain are grouped together — for example, user-events might include UserCreated, UserUpdated, and UserDeleted events. Grouping related messages in the same topic makes it easier to maintain ordering guarantees and process events in context.
But handling multiple message types introduces complexity. The consumer now needs to distinguish between different types of messages and deserialize them correctly.
Why Use Multiple Message Types in One Topic?
Some common reasons to use a single topic for multiple message types:
- Ordering Guarantees: Events for a specific key (e.g.,
UserId) are guaranteed to be delivered in order. - Logical Grouping: All user-related events are in one place, making it easier to replay and audit.
- Topic Management: Fewer topics to maintain across environments.
But this also means your consumer logic must know how to identify and handle each event type properly.
Patterns for Handling Multiple Message Types
There are two common ways to distinguish between message types:
- Type Discriminator Field Each Avro schema includes a field like
"EventType": "UserCreated"to indicate the message type. You deserialize the message into a generic base type first, then switch based on the discriminator. - Union Schemas or Topic Wrappers Define a top-level Avro schema that wraps multiple possible message types using Avro’s union support. This is harder to maintain across teams, so many choose the first option for flexibility.
Example: Deserializing and Processing Different Message Types
We’ll use option 1, where each message includes an EventType string field.
{
"type": "record",
"name": "UserEvent",
"namespace": "KafkaSchemas",
"fields": [
{ "name": "EventType", "type": "string" },
{ "name": "Payload", "type": "string" } // JSON string of actual event data
]
}
Then, when we need to deserialize the message, we can run the correct logic based on the event type:
var userEvent = result.Message.Value;
switch (userEvent.EventType)
{
case "UserCreated":
var created = JsonSerializer.Deserialize<UserCreated>(userEvent.Payload);
HandleUserCreated(created);
break;
case "UserUpdated":
var updated = JsonSerializer.Deserialize<UserUpdated>(userEvent.Payload);
HandleUserUpdated(updated);
break;
case "UserDeleted":
var deleted = JsonSerializer.Deserialize<UserDeleted>(userEvent.Payload);
HandleUserDeleted(deleted);
break;
default:
_logger.LogWarning("Unknown event type: {EventType}", userEvent.EventType);
break;
}
Best Practices and Recommendations
Error Handling Strategies in Production
- Fail Fast on Critical Errors If deserialization consistently fails or schema mismatches occur, don’t retry endlessly. Route the message to a DLQ and alert. You can’t recover from a broken schema at runtime.
- Use Retry Policies for Transient Failures Wrap business logic with retry mechanisms (like Polly) to recover from timeouts, deadlocks, or downstream service hiccups. Use exponential backoff with a cap to avoid flooding dependencies.
- Implement a Dead Letter Queue (DLQ) After N retries, send the failed message to a DLQ Kafka topic. This allows you to keep processing good messages without dropping data. Later, you can reprocess DLQ messages manually or with a batch job.
- Add Context to DLQ Messages Include the exception message and timestamp when sending to the DLQ. This makes debugging much easier.
- Guard Against Poison Messages Don’t let one bad message bring down your service. Always catch and log exceptions — isolate failures to individual messages.
Recommendations for Message Handling Efficiency
- Keep Message Sizes Small Kafka handles small messages better. If your payloads are large (e.g., over 1MB), consider storing the data elsewhere (e.g., S3) and passing a reference in the Kafka message.
- Batch Processing When possible, consume messages in batches (
Consume(CancellationToken)in a loop or using a custom poll batch), especially for high-throughput scenarios. - Avoid Unnecessary Deserialization If your consumer is filtering messages by key or headers, apply the filter before deserialization to save CPU cycles.
Conclusion
Kafka is powerful, but using it effectively in real-world .NET applications takes more than just sending and receiving messages. By integrating Schema Registry, using Avro for serialization, handling errors with retry logic and DLQs, and managing multiple message types within the same topic — you build a messaging system that’s not only functional, but also resilient and maintainable.
Cover Photo by Ivan Samkov