Confluent.Kafka
2.5.3
dotnet add package Confluent.Kafka --version 2.5.3
NuGet\Install-Package Confluent.Kafka -Version 2.5.3
<PackageReference Include="Confluent.Kafka" Version="2.5.3" />
paket add Confluent.Kafka --version 2.5.3
#r "nuget: Confluent.Kafka, 2.5.3"
// Install Confluent.Kafka as a Cake Addin #addin nuget:?package=Confluent.Kafka&version=2.5.3 // Install Confluent.Kafka as a Cake Tool #tool nuget:?package=Confluent.Kafka&version=2.5.3
Confluent's .NET Client for Apache Kafka<sup>TM</sup>
confluent-kafka-dotnet is Confluent's .NET client for Apache Kafka and the Confluent Platform.
Features:
High performance - confluent-kafka-dotnet is a lightweight wrapper around librdkafka, a finely tuned C client.
Reliability - There are a lot of details to get right when writing an Apache Kafka client. We get them right in one place (librdkafka) and leverage this work across all of our clients (also confluent-kafka-python and confluent-kafka-go).
Supported - Commercial support is offered by Confluent.
Future proof - Confluent, founded by the original creator/co-creator of Kafka, is building a streaming platform with Apache Kafka at its core. It's high priority for us that client features keep pace with core Apache Kafka and components of the Confluent Platform.
confluent-kafka-dotnet is derived from Andreas Heider's rdkafka-dotnet. We're fans of his work and were very happy to have been able to leverage rdkafka-dotnet as the basis of this client. Thanks Andreas!
Referencing
confluent-kafka-dotnet is distributed via NuGet. We provide the following packages:
- Confluent.Kafka [net462, netstandard1.3, netstandard2.0] - The core client library.
- Confluent.SchemaRegistry.Serdes.Avro [netstandard2.0] - Provides a serializer and deserializer for working with Avro serialized data with Confluent Schema Registry integration.
- Confluent.SchemaRegistry.Serdes.Protobuf [netstandard2.0] - Provides a serializer and deserializer for working with Protobuf serialized data with Confluent Schema Registry integration.
- Confluent.SchemaRegistry.Serdes.Json [netstandard2.0] - Provides a serializer and deserializer for working with Json serialized data with Confluent Schema Registry integration.
- Confluent.SchemaRegistry [netstandard2.0] - Confluent Schema Registry client (a dependency of the Confluent.SchemaRegistry.Serdes packages).
- Confluent.SchemaRegistry.Encryption [netcoreapp3.1, net6.0] - Confluent Schema Registry client-side field-level encryption client (a dependency of the other Confluent.SchemaRegistry.Encryption.* packages).
- Confluent.SchemaRegistry.Encryption.Aws [netcoreapp3.1, net6.0] - Confluent Schema Registry client-side field-level encryption client for AWS KMS.
- Confluent.SchemaRegistry.Encryption.Azure [netcoreapp3.1, net6.0] - Confluent Schema Registry client-side field-level encryption client for Azure Key Vault.
- Confluent.SchemaRegistry.Encryption.Gcp [netcoreapp3.1, net6.0] - Confluent Schema Registry client-side field-level encryption client for Google Cloud KMS.
- Confluent.SchemaRegistry.Encryption.HcVault [netcoreapp3.1, net6.0] - Confluent Schema Registry client-side field-level encryption client for Hashicorp Vault.
- Confluent.SchemaRegistry.Rules [net6.0] - Confluent Schema Registry client-side support for data quality rules (via the Common Expression Language) and schema migration rules (via JSONata).
To install Confluent.Kafka from within Visual Studio, search for Confluent.Kafka in the NuGet Package Manager UI, or run the following command in the Package Manager Console:
Install-Package Confluent.Kafka -Version 2.5.3
To add a reference to a dotnet core project, execute the following at the command line:
dotnet add package -v 2.5.3 Confluent.Kafka
Note: Confluent.Kafka
depends on the librdkafka.redist
package which provides a number of different builds of librdkafka
that are compatible with common platforms. If you are on one of these platforms this will all work seamlessly (and you don't need to explicitly reference librdkafka.redist
). If you are on a different platform, you may need to build librdkafka manually (or acquire it via other means) and load it using the Library.Load method.
Branch builds
Nuget packages corresponding to all commits to release branches are available from the following nuget package source (Note: this is not a web URL - you should specify it in the nuget package manager): https://ci.appveyor.com/nuget/confluent-kafka-dotnet. The version suffix of these nuget packages matches the appveyor build number. You can see which commit a particular build number corresponds to by looking at the AppVeyor build history
Usage
For a step-by-step guide and code samples, see Getting Started with Apache Kafka and .NET on Confluent Developer.
You can also take the free self-paced training course Apache Kafka for .NET Developers on Confluent Developer.
Take a look in the examples directory and at the integration tests for further examples.
For an overview of configuration properties, refer to the librdkafka documentation.
Basic Producer Examples
You should use the ProduceAsync
method if you would like to wait for the result of your produce
requests before proceeding. You might typically want to do this in highly concurrent scenarios,
for example in the context of handling web requests. Behind the scenes, the client will manage
optimizing communication with the Kafka brokers for you, batching requests as appropriate.
using System;
using System.Threading.Tasks;
using Confluent.Kafka;
class Program
{
public static async Task Main(string[] args)
{
var config = new ProducerConfig { BootstrapServers = "localhost:9092" };
// If serializers are not specified, default serializers from
// `Confluent.Kafka.Serializers` will be automatically used where
// available. Note: by default strings are encoded as UTF8.
using (var p = new ProducerBuilder<Null, string>(config).Build())
{
try
{
var dr = await p.ProduceAsync("test-topic", new Message<Null, string> { Value = "test" });
Console.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
}
catch (ProduceException<Null, string> e)
{
Console.WriteLine($"Delivery failed: {e.Error.Reason}");
}
}
}
}
Note that a server round-trip is slow (3ms at a minimum; actual latency depends on many factors).
In highly concurrent scenarios you will achieve high overall throughput out of the producer using
the above approach, but there will be a delay on each await
call. In stream processing
applications, where you would like to process many messages in rapid succession, you would typically
use the Produce
method instead:
using System;
using Confluent.Kafka;
class Program
{
public static void Main(string[] args)
{
var conf = new ProducerConfig { BootstrapServers = "localhost:9092" };
Action<DeliveryReport<Null, string>> handler = r =>
Console.WriteLine(!r.Error.IsError
? $"Delivered message to {r.TopicPartitionOffset}"
: $"Delivery Error: {r.Error.Reason}");
using (var p = new ProducerBuilder<Null, string>(conf).Build())
{
for (int i = 0; i < 100; ++i)
{
p.Produce("my-topic", new Message<Null, string> { Value = i.ToString() }, handler);
}
// wait for up to 10 seconds for any inflight messages to be delivered.
p.Flush(TimeSpan.FromSeconds(10));
}
}
}
Basic Consumer Example
using System;
using System.Threading;
using Confluent.Kafka;
class Program
{
public static void Main(string[] args)
{
var conf = new ConsumerConfig
{
GroupId = "test-consumer-group",
BootstrapServers = "localhost:9092",
// Note: The AutoOffsetReset property determines the start offset in the event
// there are not yet any committed offsets for the consumer group for the
// topic/partitions of interest. By default, offsets are committed
// automatically, so in this example, consumption will only start from the
// earliest message in the topic 'my-topic' the first time you run the program.
AutoOffsetReset = AutoOffsetReset.Earliest
};
using (var c = new ConsumerBuilder<Ignore, string>(conf).Build())
{
c.Subscribe("my-topic");
CancellationTokenSource cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => {
// Prevent the process from terminating.
e.Cancel = true;
cts.Cancel();
};
try
{
while (true)
{
try
{
var cr = c.Consume(cts.Token);
Console.WriteLine($"Consumed message '{cr.Value}' at: '{cr.TopicPartitionOffset}'.");
}
catch (ConsumeException e)
{
Console.WriteLine($"Error occured: {e.Error.Reason}");
}
}
}
catch (OperationCanceledException)
{
// Ensure the consumer leaves the group cleanly and final offsets are committed.
c.Close();
}
}
}
}
IHostedService and Web Application Integration
The Web example demonstrates how to integrate
Apache Kafka with a web application, including how to implement IHostedService
to realize a long running consumer poll loop, how to
register a producer as a singleton service, and how to bind configuration from an injected IConfiguration
instance.
Exactly Once Processing
The .NET Client has full support for transactions and idempotent message production, allowing you to write horizontally scalable stream processing applications with exactly once semantics. The ExactlyOnce example demonstrates this capability by way of an implementation of the classic "word count" problem, also demonstrating how to use the FASTER Key/Value store (similar to RocksDb) to materialize working state that may be larger than available memory, and incremental rebalancing to avoid stop-the-world rebalancing operations and unnecessary reloading of state when you add or remove processing nodes.
Schema Registry Integration
The three "Serdes" packages provide serializers and deserializers for Avro, Protobuf and JSON with Confluent Schema Registry integration. The Confluent.SchemaRegistry
nuget package provides a client for interfacing with
Schema Registry's REST API.
Note: All three serialization formats are supported across Confluent Platform. They each make different tradeoffs, and you should use the one that best matches to your requirements. Avro is well suited to the streaming data use-case, but the quality and maturity of the non-Java implementations lags that of Java - this is an important consideration. Protobuf and JSON both have great support in .NET.
Error Handling
Errors delivered to a client's error handler should be considered informational except when the IsFatal
flag
is set to true
, indicating that the client is in an un-recoverable state. Currently, this can only happen on
the producer, and only when enable.idempotence
has been set to true
. In all other scenarios, clients will
attempt to recover from all errors automatically.
Although calling most methods on the clients will result in a fatal error if the client is in an un-recoverable state, you should generally only need to explicitly check for fatal errors in your error handler, and handle this scenario there.
Producer
When using Produce
, to determine whether a particular message has been successfully delivered to a cluster,
check the Error
field of the DeliveryReport
during the delivery handler callback.
When using ProduceAsync
, any delivery result other than NoError
will cause the returned Task
to be in the
faulted state, with the Task.Exception
field set to a ProduceException
containing information about the message
and error via the DeliveryResult
and Error
fields. Note: if you await
the call, this means a ProduceException
will be thrown.
Consumer
All Consume
errors will result in a ConsumeException
with further information about the error and context
available via the Error
and ConsumeResult
fields.
3rd Party
There are numerous libraries that expand on the capabilities provided by Confluent.Kafka, or use Confluent.Kafka to integrate with Kafka. For more information, refer to the 3rd Party Libraries page.
Confluent Cloud
For a step-by-step guide on using the .NET client with Confluent Cloud see Getting Started with Apache Kafka and .NET on Confluent Developer.
You can also refer to the Confluent Cloud example which demonstrates how to configure the .NET client for use with Confluent Cloud.
Developer Notes
Instructions on building and testing confluent-kafka-dotnet can be found here.
Copyright (c) 2016-2019 Confluent Inc. 2015-2016 Andreas Heider
KAFKA is a registered trademark of The Apache Software Foundation and has been licensed for use by confluent-kafka-dotnet. confluent-kafka-dotnet has no affiliation with and is not endorsed by The Apache Software Foundation.
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 was computed. 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. |
.NET Core | netcoreapp1.0 was computed. netcoreapp1.1 was computed. netcoreapp2.0 was computed. netcoreapp2.1 was computed. netcoreapp2.2 was computed. netcoreapp3.0 was computed. netcoreapp3.1 was computed. |
.NET Standard | netstandard1.3 is compatible. netstandard1.4 was computed. netstandard1.5 was computed. netstandard1.6 was computed. netstandard2.0 is compatible. netstandard2.1 was computed. |
.NET Framework | net46 was computed. net461 was computed. net462 is compatible. 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 | tizen30 was computed. tizen40 was computed. tizen60 was computed. |
Universal Windows Platform | uap was computed. uap10.0 was computed. |
Xamarin.iOS | xamarinios was computed. |
Xamarin.Mac | xamarinmac was computed. |
Xamarin.TVOS | xamarintvos was computed. |
Xamarin.WatchOS | xamarinwatchos was computed. |
-
.NETFramework 4.6.2
- librdkafka.redist (>= 2.5.3)
- System.Memory (>= 4.5.0)
-
.NETStandard 1.3
- librdkafka.redist (>= 2.5.3)
- NETStandard.Library (>= 1.6.1)
- System.Console (>= 4.3.0)
- System.Linq (>= 4.3.0)
- System.Memory (>= 4.5.0)
- System.Runtime.Extensions (>= 4.3.0)
- System.Runtime.InteropServices (>= 4.3.0)
- System.Threading (>= 4.3.0)
-
.NETStandard 2.0
- librdkafka.redist (>= 2.5.3)
- System.Memory (>= 4.5.0)
-
net6.0
- librdkafka.redist (>= 2.5.3)
NuGet packages (678)
Showing the top 5 NuGet packages that depend on Confluent.Kafka:
Package | Downloads |
---|---|
Confluent.SchemaRegistry
A .NET Client for Confluent Schema Registry |
|
Confluent.SchemaRegistry.Serdes.Avro
Provides an Avro Serializer and Deserializer for use with Confluent.Kafka with Confluent Schema Registry integration |
|
AspNetCore.HealthChecks.Kafka
HealthChecks.Kafka is the health check package for Kafka. |
|
Confluent.SchemaRegistry.Serdes.Protobuf
Provides a Protobuf Serializer and Deserializer for use with Confluent.Kafka with Confluent Schema Registry integration |
|
Confluent.SchemaRegistry.Serdes.Json
Provides a JSON Serializer and Deserializer for use with Confluent.Kafka with Confluent Schema Registry integration |
GitHub repositories (72)
Showing the top 5 popular GitHub repositories that depend on Confluent.Kafka:
Repository | Stars |
---|---|
abpframework/abp
Open-source web application framework for ASP.NET Core! Offers an opinionated architecture to build enterprise software solutions with best practices on top of the .NET. Provides the fundamental infrastructure, cross-cutting-concern implementations, startup templates, application modules, UI themes, tooling and documentation.
|
|
MassTransit/MassTransit
Distributed Application Framework for .NET
|
|
dotnetcore/CAP
Distributed transaction solution in micro-service base on eventually consistency, also an eventbus with Outbox pattern
|
|
anjoy8/Blog.Core
💖 ASP.NET Core 8.0 全家桶教程,前后端分离后端接口,vue教程姊妹篇,官方文档:
|
|
dotnetcore/FreeSql
🦄 .NET aot orm, C# orm, VB.NET orm, Mysql orm, Postgresql orm, SqlServer orm, Oracle orm, Sqlite orm, Firebird orm, 达梦 orm, 人大金仓 orm, 神通 orm, 翰高 orm, 南大通用 orm, 虚谷 orm, 国产 orm, Clickhouse orm, QuestDB orm, MsAccess orm.
|
Version | Downloads | Last updated |
---|---|---|
2.5.3 | 259,143 | 9/3/2024 |
2.5.3-RC1 | 411 | 9/2/2024 |
2.5.2 | 410,592 | 8/7/2024 |
2.5.1 | 92,324 | 8/1/2024 |
2.5.0 | 520,349 | 7/11/2024 |
2.5.0-RC5 | 356 | 7/10/2024 |
2.5.0-RC4 | 324 | 7/10/2024 |
2.5.0-RC3 | 269 | 7/10/2024 |
2.4.0 | 1,984,201 | 5/7/2024 |
2.4.0-RC2 | 316 | 5/7/2024 |
2.3.0 | 8,002,397 | 10/25/2023 |
2.3.0-RC4 | 4,673 | 10/25/2023 |
2.3.0-RC3 | 3,370 | 10/23/2023 |
2.2.0 | 5,615,064 | 7/12/2023 |
2.2.0-RC2 | 3,370 | 7/12/2023 |
2.1.1 | 4,477,843 | 5/4/2023 |
2.1.1-RC1 | 4,048 | 4/26/2023 |
2.1.0 | 1,696,076 | 4/6/2023 |
2.1.0-RC3 | 3,238 | 4/5/2023 |
2.0.2 | 5,023,575 | 1/23/2023 |
1.9.4-RC1 | 151,496 | 9/21/2022 |
1.9.3 | 11,851,913 | 9/12/2022 |
1.9.2 | 4,391,228 | 8/2/2022 |
1.9.0 | 4,228,195 | 6/16/2022 |
1.8.2 | 13,202,721 | 10/19/2021 |
1.8.1 | 1,099,275 | 9/20/2021 |
1.8.0 | 316,956 | 9/16/2021 |
1.7.0 | 6,828,767 | 5/19/2021 |
1.7.0-RC4 | 9,775 | 4/23/2021 |
1.6.3 | 7,450,259 | 3/19/2021 |
1.6.2 | 1,395,554 | 2/25/2021 |
1.5.3 | 3,149,770 | 12/9/2020 |
1.5.2 | 3,896,210 | 10/20/2020 |
1.5.1 | 2,285,560 | 8/20/2020 |
1.5.0 | 1,689,387 | 7/21/2020 |
1.4.4 | 2,081,325 | 6/29/2020 |
1.4.3 | 2,401,137 | 5/19/2020 |
1.4.2 | 974,230 | 5/6/2020 |
1.4.0 | 1,793,320 | 4/3/2020 |
1.3.0 | 4,993,293 | 12/4/2019 |
1.2.2 | 704,036 | 11/14/2019 |
1.2.1 | 1,006,839 | 10/9/2019 |
1.2.0 | 836,528 | 9/19/2019 |
1.1.0 | 1,702,311 | 6/27/2019 |
1.0.1.1 | 1,116,037 | 6/3/2019 |
1.0.1 | 193,939 | 5/28/2019 |
1.0.0 | 545,702 | 4/25/2019 |
1.0.0-beta3 | 122,744 | 2/7/2019 |
1.0.0-beta2 | 655,624 | 10/15/2018 |
1.0.0-beta | 96,417 | 9/24/2018 |
0.11.6 | 1,747,527 | 10/19/2018 |
0.11.5 | 616,150 | 7/19/2018 |
0.11.4 | 846,827 | 3/29/2018 |
0.11.3 | 571,857 | 12/4/2017 |
0.11.2 | 172,790 | 10/24/2017 |
0.11.1 | 19,789 | 10/18/2017 |
0.11.0 | 188,165 | 7/19/2017 |
0.9.5 | 153,471 | 4/21/2017 |
0.9.4 | 29,959 | 3/1/2017 |