Confluent.Kafka.DependencyInjection
3.4.0
dotnet add package Confluent.Kafka.DependencyInjection --version 3.4.0
NuGet\Install-Package Confluent.Kafka.DependencyInjection -Version 3.4.0
<PackageReference Include="Confluent.Kafka.DependencyInjection" Version="3.4.0" />
<PackageVersion Include="Confluent.Kafka.DependencyInjection" Version="3.4.0" />
<PackageReference Include="Confluent.Kafka.DependencyInjection" />
paket add Confluent.Kafka.DependencyInjection --version 3.4.0
#r "nuget: Confluent.Kafka.DependencyInjection, 3.4.0"
#:package Confluent.Kafka.DependencyInjection@3.4.0
#addin nuget:?package=Confluent.Kafka.DependencyInjection&version=3.4.0
#tool nuget:?package=Confluent.Kafka.DependencyInjection&version=3.4.0
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
.
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, a common pattern for consuming is to implement a .NET BackgroundService
.
class MyWorker(IConsumer<Ignore, MyType> consumer) : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("Consumer service started.");
// ConsumeAllAsync() is an extension provided by this library.
await foreach (var result in consumer.ConsumeAllAsync().WithCancellation(stoppingToken))
{
Console.WriteLine($"Message {result.TopicPartitionOffset} processed.");
}
}
}
Register the service with the container.
services.AddHostedService<MyWorker>();
Product | Versions Compatible and additional computed target framework versions. |
---|---|
.NET | net5.0 was computed. net5.0-windows was computed. net6.0 is compatible. net6.0-android was computed. net6.0-ios was computed. net6.0-maccatalyst was computed. net6.0-macos was computed. net6.0-tvos was computed. net6.0-windows was computed. net7.0 was computed. net7.0-android was computed. net7.0-ios was computed. net7.0-maccatalyst was computed. net7.0-macos was computed. net7.0-tvos was computed. net7.0-windows was computed. net8.0 is compatible. net8.0-android was computed. net8.0-browser was computed. net8.0-ios was computed. net8.0-maccatalyst was computed. net8.0-macos was computed. net8.0-tvos was computed. net8.0-windows was computed. net9.0 was computed. net9.0-android was computed. net9.0-browser was computed. net9.0-ios was computed. net9.0-maccatalyst was computed. net9.0-macos was computed. net9.0-tvos was computed. net9.0-windows was computed. net10.0 was computed. net10.0-android was computed. net10.0-browser was computed. net10.0-ios was computed. net10.0-maccatalyst was computed. net10.0-macos was computed. net10.0-tvos was computed. net10.0-windows was computed. |
.NET Core | netcoreapp2.0 was computed. netcoreapp2.1 was computed. netcoreapp2.2 was computed. netcoreapp3.0 was computed. netcoreapp3.1 was computed. |
.NET Standard | netstandard2.0 is compatible. netstandard2.1 was computed. |
.NET Framework | net461 was computed. net462 was computed. net463 was computed. net47 was computed. net471 was computed. net472 was computed. net48 was computed. net481 was computed. |
MonoAndroid | monoandroid was computed. |
MonoMac | monomac was computed. |
MonoTouch | monotouch was computed. |
Tizen | tizen40 was computed. tizen60 was computed. |
Xamarin.iOS | xamarinios was computed. |
Xamarin.Mac | xamarinmac was computed. |
Xamarin.TVOS | xamarintvos was computed. |
Xamarin.WatchOS | xamarinwatchos was computed. |
-
.NETStandard 2.0
- Confluent.Kafka (>= 2.11.0)
- Microsoft.Extensions.Configuration.Abstractions (>= 8.0.0)
- Microsoft.Extensions.DependencyInjection.Abstractions (>= 8.0.2)
- Microsoft.Extensions.Logging.Abstractions (>= 8.0.3)
- Microsoft.Extensions.Options (>= 8.0.2)
-
net6.0
- Confluent.Kafka (>= 2.11.0)
- Microsoft.Extensions.Configuration.Abstractions (>= 8.0.0)
- Microsoft.Extensions.DependencyInjection.Abstractions (>= 8.0.2)
- Microsoft.Extensions.Logging.Abstractions (>= 8.0.3)
- Microsoft.Extensions.Options (>= 8.0.2)
-
net8.0
- Confluent.Kafka (>= 2.11.0)
- Microsoft.Extensions.Configuration.Abstractions (>= 8.0.0)
- Microsoft.Extensions.DependencyInjection.Abstractions (>= 8.0.2)
- Microsoft.Extensions.Logging.Abstractions (>= 8.0.3)
- Microsoft.Extensions.Options (>= 8.0.2)
NuGet packages (2)
Showing the top 2 NuGet packages that depend on Confluent.Kafka.DependencyInjection:
Package | Downloads |
---|---|
DanielXOO.Serilog.Sinks.Kafka
Serilog sink for kafka |
|
SkillAssessor.Common.Logging
Package Description |
GitHub repositories
This package is not used by any popular GitHub repositories.
Version | Downloads | Last Updated |
---|---|---|
3.4.0 | 28 | 6 days ago |
3.3.0 | 10,255 | a month ago |
3.2.0 | 35,885 | 4 months ago |
3.1.0 | 280,827 | 7/21/2023 |
3.0.2 | 263 | 7/21/2023 |
3.0.1 | 14,941 | 4/16/2023 |
3.0.0 | 60,106 | 3/28/2023 |
3.0.0-test3 | 220 | 3/27/2023 |
3.0.0-test2 | 644 | 3/21/2023 |
3.0.0-test | 215 | 3/10/2023 |
2.2.0 | 17,227 | 2/17/2023 |
2.1.2 | 11,317 | 2/2/2023 |
2.1.1 | 22,055 | 1/3/2023 |
2.1.0 | 31,037 | 11/3/2022 |
2.0.1 | 2,484 | 11/2/2022 |
2.0.1-test | 217 | 11/2/2022 |
2.0.0 | 127,130 | 6/21/2021 |
1.1.0 | 7,425 | 2/28/2021 |
1.0.0 | 844 | 1/8/2021 |
0.1.0 | 974 | 6/15/2020 |