Confluent.Kafka.DependencyInjection
3.3.0
dotnet add package Confluent.Kafka.DependencyInjection --version 3.3.0
NuGet\Install-Package Confluent.Kafka.DependencyInjection -Version 3.3.0
<PackageReference Include="Confluent.Kafka.DependencyInjection" Version="3.3.0" />
<PackageVersion Include="Confluent.Kafka.DependencyInjection" Version="3.3.0" />
<PackageReference Include="Confluent.Kafka.DependencyInjection" />
paket add Confluent.Kafka.DependencyInjection --version 3.3.0
#r "nuget: Confluent.Kafka.DependencyInjection, 3.3.0"
#addin nuget:?package=Confluent.Kafka.DependencyInjection&version=3.3.0
#tool nuget:?package=Confluent.Kafka.DependencyInjection&version=3.3.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>();
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.10.1)
- 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.10.1)
- 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.10.1)
- 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.3.0 | 5,318 | 17 days ago |
3.2.0 | 35,290 | 4 months ago |
3.1.0 | 278,945 | 7/21/2023 |
3.0.2 | 254 | 7/21/2023 |
3.0.1 | 14,887 | 4/16/2023 |
3.0.0 | 60,004 | 3/28/2023 |
3.0.0-test3 | 210 | 3/27/2023 |
3.0.0-test2 | 633 | 3/21/2023 |
3.0.0-test | 209 | 3/10/2023 |
2.2.0 | 17,191 | 2/17/2023 |
2.1.2 | 10,942 | 2/2/2023 |
2.1.1 | 21,981 | 1/3/2023 |
2.1.0 | 30,888 | 11/3/2022 |
2.0.1 | 2,361 | 11/2/2022 |
2.0.1-test | 207 | 11/2/2022 |
2.0.0 | 126,538 | 6/21/2021 |
1.1.0 | 7,416 | 2/28/2021 |
1.0.0 | 834 | 1/8/2021 |
0.1.0 | 964 | 6/15/2020 |