confluent-kafka-dotnet

Confluent’s .NET Client for Apache Kafka®

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 [netstandard2.0, net462, net8.0, net10.0] - The core client library.
• Confluent.Kafka.OAuthBearer.Aws [netstandard2.1, net462, net8.0, net10.0] - Optional AWS IAM-based OAUTHBEARER authentication for Confluent.Kafka.
• Confluent.SchemaRegistry.Serdes.Avro [netstandard2.0, net462, net8.0, net10.0] - Provides a serializer and deserializer for working with Avro serialized data with Confluent Schema Registry integration.
• Confluent.SchemaRegistry.Serdes.Protobuf [netstandard2.0, net462, net8.0, net10.0] - Provides a serializer and deserializer for working with Protobuf serialized data with Confluent Schema Registry integration.
• Confluent.SchemaRegistry.Serdes.Json [netstandard2.0, net462, net8.0, net10.0] - Provides a serializer and deserializer for working with Json serialized data with Confluent Schema Registry integration.
• Confluent.SchemaRegistry [netstandard2.0, net462, net8.0, net10.0] - Confluent Schema Registry client (a dependency of the Confluent.SchemaRegistry.Serdes packages).
• Confluent.SchemaRegistry.Encryption [netstandard2.1, net462, net8.0, net10.0] - Confluent Schema Registry client-side field-level encryption client (a dependency of the other Confluent.SchemaRegistry.Encryption.* packages).
• Confluent.SchemaRegistry.Encryption.Aws [netstandard2.1, net462, net8.0, net10.0] - Confluent Schema Registry client-side field-level encryption client for AWS KMS.
• Confluent.SchemaRegistry.Encryption.Azure [netstandard2.1, net462, net8.0, net10.0] - Confluent Schema Registry client-side field-level encryption client for Azure Key Vault.
• Confluent.SchemaRegistry.Encryption.Gcp [netstandard2.1, net462, net8.0, net10.0] - Confluent Schema Registry client-side field-level encryption client for Google Cloud KMS.
• Confluent.SchemaRegistry.Encryption.HcVault [netstandard2.1, net462, net8.0, net10.0] - Confluent Schema Registry client-side field-level encryption client for Hashicorp Vault.
• Confluent.SchemaRegistry.Encryption.AliCloud [netstandard2.1, net462, net8.0, net10.0] - Confluent Schema Registry client-side field-level encryption client for Alibaba Cloud KMS.
• Confluent.SchemaRegistry.Rules [netstandard2.1, net462, net8.0, net10.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.16.0

To add a reference to a dotnet core project, execute the following at the command line:

dotnet add package -v 2.16.0 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.

s390x (IBM Z)

linux-s390x is supported: the managed assemblies are architecture independent, and librdkafka.redist ships an s390x native library. The APIs, configuration and wire behaviour are identical to any other platform. A few environment constraints are specific to s390x:

• Use a Red Hat distributed .NET runtime. Microsoft does not publish a .NET runtime for s390x. Red Hat does, for .NET 8/9/10, via RHEL/UBI RPMs and the registry.access.redhat.com/ubi9/dotnet-* container images.
• glibc only. There is no Alpine/musl build of librdkafka for s390x.
• No NativeAOT, single file apps, or ReadyToRun (R2R), per Red Hat’s documented .NET limitations on s390x. None of these affect normal Confluent.Kafka usage.
• No GSSAPI/Kerberos. The s390x librdkafka build is configured with --disable-gssapi. Other SASL mechanisms are unaffected.
• Set MONO_THREADS_SUSPEND=coop if the application uses a log handler (SetLogHandler) or a custom partitioner. .NET on s390x runs on the Mono runtime, and in its default mode the process can occasionally hang during garbage collection when these callbacks run. Set the environment variable before the process starts (for example in the Dockerfile or service definition). The value must be exactly coop: Mono aborts on an invalid or empty value.

Building this repository from source on s390x has two wrinkles, neither of which affects consumers of the published packages (see scripts/run-tests-s390x.sh):

• Build the libraries and test projects individually rather than restoring the whole solution. The executable helper projects reference the portable Microsoft.NETCore.App.Host.linux-s390x app host, which Microsoft does not publish, so a solution-wide restore fails on it. Red Hat’s SDK ships a rhel.9-s390x app host, so the libraries and test projects themselves build and run natively.
• When building on RHEL/UBI, allow SHA-1 signatures so that strong naming can run.

Branch builds

Nuget packages corresponding to commits to branches are produced as CI artifacts by Semaphore, with a version suffix of the form ci-<job-id>.

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.

Production applications should serialize with Schema Registry. Producing plain string or raw values leads to data-quality issues, broken consumers, and ungovernable data. The producer examples below require the Confluent.SchemaRegistry and Confluent.SchemaRegistry.Serdes.Json packages in addition to Confluent.Kafka.

using System;
using System.Threading.Tasks;
using Confluent.Kafka;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;

class Program
{
    public class User
    {
        public string Name { get; set; }
        public int FavoriteNumber { get; set; }
    }

    public static async Task Main(string[] args)
    {
        var config = new ProducerConfig { BootstrapServers = "localhost:9092" };
        var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081" };

        // The JSON schema is generated from the User type, then registered and
        // validated on produce.
        using var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
        using var p = new ProducerBuilder<Null, User>(config)
            .SetValueSerializer(new JsonSerializer<User>(schemaRegistry))
            .Build();

        try
        {
            var user = new User { Name = "Confluent", FavoriteNumber = 42 };
            var dr = await p.ProduceAsync("my-topic", new Message<Null, User> { Value = user });
            Console.WriteLine($"Delivered to '{dr.TopicPartitionOffset}'");
        }
        catch (ProduceException<Null, User> e)
        {
            Console.WriteLine($"Delivery failed: {e.Error.Reason}");
        }
    }
}

Avro and Protobuf serializers are also available — see Schema Registry Integration and the JsonSerialization, AvroSpecific, and Protobuf examples.

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;
using Confluent.Kafka.SyncOverAsync;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;

class Program
{
    public class User
    {
        public string Name { get; set; }
        public int FavoriteNumber { get; set; }
    }

    public static void Main(string[] args)
    {
        var conf = new ProducerConfig { BootstrapServers = "localhost:9092" };
        var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081" };

        Action<DeliveryReport<Null, User>> handler = r =>
            Console.WriteLine(!r.Error.IsError
                ? $"Delivered message to {r.TopicPartitionOffset}"
                : $"Delivery Error: {r.Error.Reason}");

        using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
        // Note: the Schema Registry serializers are asynchronous. The fire-and-forget
        // `Produce` method requires a synchronous serializer, so wrap it with
        // `AsSyncOverAsync`. Schema lookups are cached, so this only blocks on the
        // first message for a given schema.
        using (var p = new ProducerBuilder<Null, User>(conf)
            .SetValueSerializer(new JsonSerializer<User>(schemaRegistry).AsSyncOverAsync())
            .Build())
        {
            for (int i = 0; i < 100; ++i)
            {
                var user = new User { Name = "Confluent", FavoriteNumber = i };
                p.Produce("my-topic", new Message<Null, User> { Value = user }, handler);
            }

            // wait for up to 10 seconds for any inflight messages to be delivered.
            p.Flush(TimeSpan.FromSeconds(10));
        }
    }
}

Basic Consumer Example

This example consumes the messages produced by the examples above. It deserializes with Schema Registry, so it requires the Confluent.SchemaRegistry and Confluent.SchemaRegistry.Serdes.Json packages in addition to Confluent.Kafka.

using System;
using System.Threading;
using Confluent.Kafka;
using Confluent.Kafka.SyncOverAsync;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;

class Program
{
    public class User
    {
        public string Name { get; set; }
        public int FavoriteNumber { get; set; }
    }

    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
        };
        var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081" };

        using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
        using (var c = new ConsumerBuilder<Ignore, User>(conf)
            .SetValueDeserializer(new JsonDeserializer<User>(schemaRegistry).AsSyncOverAsync())
            .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 '{cr.Message.Value.Name}' ({cr.Message.Value.FavoriteNumber}) 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 © 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.