Kafka Dependency Injection

An extension of Confluent's Kafka client for use with Microsoft.Extensions.DependencyInjection (and friends).

Features

• Configure/resolve Kafka clients using the service container.
• Load client config properties using Microsoft.Extensions.Configuration.
• Automatically log client events using Microsoft.Extensions.Logging.
• Extend a base hosted service to consume/process Kafka messages with Microsoft.Extensions.Hosting.

Installation

Add the NuGet package to your project:

$ dotnet add package Confluent.Kafka.DependencyInjection

Usage

Resolving clients

Kafka DI works out-of-the-box after registering services with an IServiceCollection.

services.AddKafkaClient(); services.AddTransient<MyService>();

Inject Kafka clients via constructor.

public MyService(IProducer<string, byte[]> producer, IConsumer<Ignore, MyType> consumer, IAdminClient adminClient) { // Clients are singletons managed by the container. Producer = producer; Consumer = consumer; AdminClient = adminClient; }

Configuring clients

Client configuration properties are bound to the Kafka section of .NET configuration providers, such as appsettings.json.

{ "Kafka": { "Producer": { "bootstrap.servers": "localhost:9092", "transactional.id": "example" }, "Consumer": { "bootstrap.servers": "localhost:9092", "group.id": "example" }, "Admin": { "bootstrap.servers": "localhost:9092" } } }

You can also specify configuration properties using the options pattern.

// Prepare consumers for manual offset storage. services.Configure<ConsumerConfig>(x => x.EnableAutoOffsetStore = false);

Configure serialization by registering the appropriate interface.

// "Open" generic registrations apply to all key/value types (except built-in types). services.AddTransient(typeof(IAsyncDeserializer<>), typeof(JsonDeserializer<>)); // Configure schema registry (required by Confluent serializers). services.AddSingleton<ISchemaRegistryClient>( x => new CachedSchemaRegistryClient(new SchemaRegistryConfig { Url = "localhost:8081" }));

For advanced scenarios, implement IClientBuilderSetup to customize clients further.

class MyClientSetup : IClientBuilderSetup { public void Apply<TKey, TValue>(ProducerBuilder<TKey, TValue> builder) { builder.SetStatisticsHandler(OnStatistics); } public void Apply<TKey, TValue>(ConsumerBuilder<TKey, TValue> builder) { builder.SetStatisticsHandler(OnStatistics); } public void Apply(AdminClientBuilder builder) { builder.SetStatisticsHandler(OnStatistics); } void OnStatistics(IClient client, string statistics) { Console.WriteLine($"New statistics available for {client.Name}"); } }

Register custom setup with services.

services.AddTransient<IClientBuilderSetup, MyClientSetup>();

Consumer hosting

Once the client is configured, implement ConsumerService to integrate with the .NET host.

class MyWorker( IConsumer<Ignore, MyType> consumer, IOptions<ConsumerHostingOptions> options, ILogger<MyWorker> logger) : ConsumerService<Ignore, MyType>(consumer, options, logger) { protected override ValueTask ProcessAsync( ConsumeResult<Ignore, MyType> result, CancellationToken cancellationToken) { // Process the message. return ValueTask.CompletedTask; } }

Register the service with the container.

builder.Services.AddHostedService<MyWorker>(); // Bind consumer hosting configuration. var hostingOptions = builder.Services.AddOptions<ConsumerHostingOptions>() .BindConfiguration("Kafka:Hosting");

Hosted consumer features:

• Configurable parallelism while preserving order for each Kafka message key.
• Parallel-safe offset storage to achieve "at least once" delivery semantics.

{ "Kafka": { "Consumer": { "bootstrap.servers": "localhost:9092", "group.id": "example", "enable.auto.offset.store": "false" }, "Hosting": { "Disabled": false, "Subscription": [ "my-topic" ], "MaxDegreeOfParallelism": 10, "StoreProcessedOffsets": true } } }