Kafka with .NET: concepts, trade-offs and production lessons
If you've built real-time systems at scale, you know that message brokers can either be your best friend or your biggest headache. Kafka with .NET: concepts, trade-offs and production lessons is what we'll explore today, drawing from experience migrating order pipelines, user event streams, and financial transaction feeds into production Kafka clusters. We'll cover the Confluent client library, show you how to handle idempotence, schemas, and failure modes that can silently corrupt your data or lose messages.
Producer basics — idempotent producer, acks=all, and a key for ordering
Let's start with the producer. When you first push data to Kafka, you face a choice: how much consistency do you want, and how much throughput are you willing to sacrifice?
In production, we typically reach for three settings. Idempotence ensures that if your producer retries a send (because the network hiccupped or a broker temporarily went down), Kafka deduplicates it on the broker side, so you don't end up with two copies of the same message. Acks=all means the broker won't confirm the write until all in-sync replicas have persisted it—slower, but safer. And a key on each message ensures ordering: all messages with the same key go to the same partition, maintaining causality for related events.
Here's how we set it up in a real order service:
var cfg = new ProducerConfig
{
BootstrapServers = "kafka:9092",
EnableIdempotence = true,
Acks = Acks.All,
Retries = int.MaxValue,
MaxInFlightRequestsPerConnection = 5,
CompressionType = CompressionType.Snappy
};
using var producer = new ProducerBuilder<string, string>(cfg)
.SetErrorHandler((_, e) =>
_logger.LogError($"Producer error: {e.Reason}"))
.Build();
var orderId = "order-12345";
var orderJson = JsonSerializer.Serialize(new {
id = orderId,
amount = 99.99,
timestamp = DateTime.UtcNow
});
try
{
var result = await producer.ProduceAsync("orders",
new Message<string, string>
{
Key = orderId,
Value = orderJson
});
_logger.LogInformation($"Sent to partition {result.Partition} at offset {result.Offset}");
}
catch (ProduceException<string, string> ex)
{
_logger.LogError($"Failed to send order: {ex.Message}");
throw;
}Notice a few things here. We set Retries = int.MaxValue so transient errors don't kill the producer. MaxInFlightRequestsPerConnection = 5 lets us pipeline requests for better throughput while still respecting ordering per partition. Snappy compression is fast and reduces bandwidth by 40–50% in our experience. The Key is the order ID, so all events for an order land in the same partition—crucial if a downstream consumer needs to maintain order state.
The trade-off: idempotence + acks=all adds latency. We measured 5–15ms per produce call versus 2–3ms without safety. In an order pipeline handling thousands of orders per second, that adds up. You'll decide based on your SLA: is a duplicate order or a lost order worse? For us, duplicates are caught at the application level (we check if an order already exists), so the safety was worth it.
Consumers in a BackgroundService — poll, process, commit
Producing is one half. Consuming—processing those messages reliably—is where most bugs hide.
We wrap our consumer in a BackgroundService, which runs alongside your ASP.NET application. The pattern is simple: poll for messages, process each one, commit the offset only after successful handling. If processing fails, we don't commit, so the message reappears on the next poll.
public class OrderConsumerService : BackgroundService
{
private readonly IConsumer<string, string> _consumer;
private readonly IOrderProcessor _processor;
private readonly ILogger<OrderConsumerService> _logger;
public OrderConsumerService(
IConsumer<string, string> consumer,
IOrderProcessor processor,
ILogger<OrderConsumerService> logger)
{
_consumer = consumer;
_processor = processor;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_consumer.Subscribe("orders");
while (!stoppingToken.IsCancellationRequested)
{
try
{
var message = _consumer.Consume(TimeSpan.FromSeconds(1));
if (message == null) continue;
_logger.LogInformation($"Processing order {message.Key}");
await _processor.HandleOrderAsync(message.Value, stoppingToken);
_consumer.Commit(message);
_logger.LogInformation($"Committed offset {message.Offset}");
}
catch (ConsumeException ex)
{
_logger.LogError($"Consume error: {ex.Error.Reason}");
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to process message, will retry on next poll");
}
}
_consumer.Close();
}
}
// In Startup/Program.cs:
services.AddSingleton(sp => new ConsumerConfig
{
BootstrapServers = "kafka:9092",
GroupId = "order-processor-group",
AutoOffsetReset = AutoOffsetReset.Earliest,
EnableAutoCommit = false,
IsolationLevel = IsolationLevel.ReadCommitted,
MaxPollIntervalMs = 300000
});
services.AddSingleton<IConsumer<string, string>>(sp =>
new ConsumerBuilder<string, string>(sp.GetRequiredService<ConsumerConfig>())
.Build());
services.AddHostedService<OrderConsumerService>();Key points. EnableAutoCommit = false means we commit manually, only after processing succeeds. IsolationLevel = IsolationLevel.ReadCommitted ensures we don't read uncommitted data from producers that crashed mid-write. MaxPollIntervalMs = 300000 gives us up to 5 minutes to process a batch of messages before the broker considers us dead and rebalances the group.
The consumer polls in a tight loop, blocking for up to 1 second waiting for new messages. When a message arrives, we process it synchronously (you can parallelize with channels or TPL Dataflow if you need). Only on success do we commit—which updates the consumer's offset for that partition in Kafka's internal __consumer_offsets topic. If the pod crashes before commit, the message reappears when we restart. It's a guarantee: you'll process every message at least once.
At least once means idempotency at the application level is mandatory. If your HandleOrderAsync is not idempotent (e.g., it increments a counter), you'll double-count. We always check WHERE order_id NOT IN (already_processed) before inserting.
Schemas — Avro or Protobuf with a schema registry
JSON is fine for prototyping, but once you're moving gigabytes of data daily, schema format matters. We migrated one order pipeline from JSON to Avro and saw payloads shrink from 400 bytes to 120 bytes per message. Over millions of messages, that's real money in storage and bandwidth.
A schema registry (Confluent's or self-hosted) stores schemas centrally. Producers and consumers register their schema version, and the broker embeds a schema ID in each message. This enforces contracts: if a consumer expects a version-3 schema and the producer sends version-4, the registry can validate compatibility.
Here's a simple Avro schema for orders:
// Define your Avro schema (JSON format)
const string avroSchema = @"
{
""type"": ""record"",
""name"": ""Order"",
""namespace"": ""com.example.orders"",
""fields"": [
{ ""name"": ""id"", ""type"": ""string"" },
{ ""name"": ""customerId"", ""type"": ""string"" },
{ ""name"": ""amount"", ""type"": ""double"" },
{ ""name"": ""createdAt"", ""type"": ""long"" }
]
}";
// Using Confluent's Avro serializer:
services.AddSingleton<IAsyncSerializer<Order>>(sp =>
{
var config = new AvroSerializerConfig
{
UseLatestVersion = true
};
return new AvroSerializer<Order>(sp.GetRequiredService<ISchemaRegistryClient>(), config);
});
// In your producer:
var producerConfig = new ProducerConfig
{
BootstrapServers = "kafka:9092",
EnableIdempotence = true
};
var producerBuilder = new ProducerBuilder<string, Order>(producerConfig)
.SetValueSerializer(sp.GetRequiredService<IAsyncSerializer<Order>>())
.Build();
var order = new Order
{
Id = "order-999",
CustomerId = "cust-123",
Amount = 149.99,
CreatedAt = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
await producerBuilder.ProduceAsync("orders", new Message<string, Order>
{
Key = order.Id,
Value = order
});Avro is more compact than JSON and Protobuf. Protobuf is language-agnostic and better if your team uses Python, Go, and Java alongside .NET. Pick based on your ecosystem and compression tolerance.
Error handling — retry topics and dead-letter topics
In production, messages will fail to process. A customer record might be corrupted, or a downstream API might be down. You need a strategy that doesn't block the partition.
We use a three-tier approach:
- Retry topic: If processing fails, send the message to a retry topic with an exponential backoff header. A separate consumer polls the retry topic every 30 seconds (or 5 minutes on second retry), attempts reprocessing, and either commits or bumps to the next retry level.
- Dead-letter topic: After N retries, send to a dead-letter queue. A team monitor watches this and manually investigates.
- Never block the partition: Always catch and handle exceptions so one bad message doesn't poison a whole consumer group.
public class RobustOrderConsumerService : BackgroundService
{
private readonly IConsumer<string, string> _consumer;
private readonly IProducer<string, string> _producer;
private readonly IOrderProcessor _processor;
private readonly ILogger<RobustOrderConsumerService> _logger;
private const int MaxRetries = 3;
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_consumer.Subscribe("orders");
while (!stoppingToken.IsCancellationRequested)
{
try
{
var message = _consumer.Consume(TimeSpan.FromSeconds(1));
if (message == null) continue;
var retryCount = GetRetryCount(message.Headers);
try
{
await _processor.HandleOrderAsync(message.Value, stoppingToken);
_consumer.Commit(message);
}
catch (Exception ex)
{
_logger.LogWarning(ex, $"Processing failed for order {message.Key}, retry {retryCount}");
if (retryCount < MaxRetries)
{
await SendToRetryTopicAsync(message, retryCount + 1);
}
else
{
await SendToDeadLetterAsync(message, ex);
}
_consumer.Commit(message); // Commit even on failure to move forward
}
}
catch (ConsumeException ex)
{
_logger.LogError(ex, "Fatal consume error");
}
}
}
private async Task SendToRetryTopicAsync(ConsumeResult<string, string> message, int nextRetry)
{
var headers = new Headers(message.Headers ?? new Headers());
headers.Add("x-retry-count", BitConverter.GetBytes(nextRetry));
headers.Add("x-original-timestamp", BitConverter.GetBytes(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()));
await _producer.ProduceAsync("orders-retry", new Message<string, string>
{
Key = message.Key,
Value = message.Value,
Headers = headers
});
_logger.LogInformation($"Sent order {message.Key} to retry topic (attempt {nextRetry})");
}
private async Task SendToDeadLetterAsync(ConsumeResult<string, string> message, Exception ex)
{
var dlqMessage = new
{
original_key = message.Key,
original_value = message.Value,
error = ex.Message,
timestamp = DateTime.UtcNow
};
await _producer.ProduceAsync("orders-dlq", new Message<string, string>
{
Key = message.Key,
Value = JsonSerializer.Serialize(dlqMessage)
});
_logger.LogError($"Order {message.Key} sent to dead-letter queue");
}
private int GetRetryCount(Headers headers)
{
var retryHeader = headers?.FirstOrDefault(h => h.Key == "x-retry-count");
if (retryHeader?.Value == null) return 0;
return BitConverter.ToInt32(retryHeader.Value.Value, 0);
}
}This approach keeps the main partition flowing. Bad messages move sideways to retry or DLQ topics, so they don't stall consumer lag on the primary topic. Your operations team gets alerts when the DLQ has messages, and they can replay or fix and reprocess at their pace.
Common mistakes
- Committing before processing. If you commit the offset and then your app crashes during processing, that message is gone. Always commit after success.
- No key or inconsistent keys. Without a key (or with a random key each time), messages scatter across partitions and ordering breaks. Be deliberate about your partitioning strategy.
- Ignoring rebalancing. When consumers join or leave a group, Kafka rebalances partitions. If your app holds long-lived state per partition, you'll lose it. Design for statelessness or use distributed state (Redis, database).
- Blocking the consumer thread. If your processing is slow (e.g., a synchronous HTTP call), the consumer will lag. Parallelize with channels or async/await, but keep failure handling intact.
- No monitoring on consumer lag. Lag is the difference between the latest offset in the topic and your consumer's committed offset. High lag means you're falling behind. Alert on it.
Key takeaways
- Enable idempotence, set acks=all, and use a key for ordering. Accept the latency trade-off; correctness is harder to debug later.
- Consume in a loop, process, then commit offsets manually. Never commit before confirming success.
- Schema registries (Avro or Protobuf) enforce contracts and shrink payloads. Worth the complexity at scale.
- Implement retry and dead-letter topics to decouple failures from the main pipeline. One bad message should never block a partition.
- Test rebalancing, lag, and failover scenarios in staging. Kafka's correctness guarantees only work if your consumer handles edge cases gracefully.
Comments
Loading comments…