<a id="dotnet-client"></a>

# .NET Client for Apache Kafka

Confluent develops and maintains [confluent-kafka-dotnet](https://github.com/confluentinc/confluent-kafka-dotnet), a .NET library that
provides a high-level producer, consumer and AdminClient compatible with all
Apache Kafka® brokers version 0.8 and later, Confluent Cloud and Confluent Platform. You can find a
[changelog](https://github.com/confluentinc/confluent-kafka-dotnet/blob/master/CHANGELOG.md)
of release updates in the GitHub client repo.

<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

<a id="installation-dotnet-client"></a>

## Installation

confluent-kafka-dotnet is made available via [NuGet](http://www.nuget.org/packages/Confluent.Kafka/). It’s a
binding to the C client librdkafka, which is provided automatically via the dependent
[librdkafka.redist](https://www.nuget.org/packages/librdkafka.redist) package for a number
of popular platforms - linux-x64, osx-arm64 (Apple Silicon), osx-x64, win-x64, and win-x86.

confluent-kafka-dotnet is compatible with the .NET Framework >= v4.6.2, .NET Core >= v1.0 and
.NET Standard >= v1.3. Mono is not officially supported.

In addition to the `Confluent.Kafka` package, we provide the `Confluent.SchemaRegistry`,
`Confluent.SchemaRegistry.Serdes.Protobuf`, `Confluent.SchemaRegistry.Serdes.Json` and
`Confluent.SchemaRegistry.Serdes.Avro` packages for integration with Confluent Schema Registry.

## Example Code

For a step-by-step tutorial using the .NET client including code samples for the producer and consumer
see [this guide](https://developer.confluent.io/get-started/dotnet/).

There are also a number of examples demonstrating various aspects of the client in the
[client github repo](https://github.com/confluentinc/confluent-kafka-dotnet/tree/master/examples)
and an additional example [here](/platform/current/tutorials/examples/clients/docs/csharp.html).

<a id="producer-dotnet-client"></a>

## Producer

To create a .NET Producer, first construct an instance of the `ProducerConfig` class,
then pass this into the `ProducerBuilder`’s constructor:

```bash
using Confluent.Kafka;
using System.Net;

...

var config = new ProducerConfig
{
    BootstrapServers = "host1:9092",
    ...
};

using (var producer = new ProducerBuilder<Null, string>(config).Build())
{
    ...
}
```

To write messages to Kafka, you can use the `ProduceAsync` method:

```bash
var result = await producer.ProduceAsync("weblog", new Message<Null, string> { Value="a log message" });
```

The `Task` returned by `ProduceAsync` will not complete until the success (or otherwise) of the request is
known. As such, `await``ing the ``ProduceAsync` call will result in very low throughput except in highly
concurrent applications such as an ASP.NET request handlers. In that case, near concurrent requests will be
automatically batched together for efficiency by the client behind the scenes.

For high throughput processing, you can utilize `Tasks.WhenAll` to `await` a block of `ProduceAsync`
requests simultaneously. Alternatively, you can use the `Produce` method:

```bash
public static void handler(DeliveryReport<Null, string>)
{
    ...
}

public static process(...)
{
    ...
    producer.Produce(
        "my-topic", new Message<Null, string> { Value = "hello world" }, handler);
}
```

The `Produce` method is also asynchronous, in that it never blocks. Message delivery
information is made available out-of-band via the (optional) delivery report handler on a background
thread. The `Produce` method maps more directly to the underlying librdkafka produce API, so
comes with a bit less overhead than `ProduceAsync`. `ProduceAsync` is still very performant
though - capable of producing hundreds of thousands of messages per second on typical hardware.

The .NET client is mature and fully featured. For in depth information on its capability, refer to the
[librdkafka documentation](https://github.com/edenhill/librdkafka/blob/master/INTRODUCTION.md).
One feature worth calling out is support for idempotent produce - simply set the `EnableIdempotence`
configuration property to `true` for exactly once, in order delivery guarantees. Another is support
for the transactional producer API. For example of how to use transactions from .NET, refer to the
[ExactlyOnce example](https://github.com/confluentinc/confluent-kafka-dotnet/tree/master/examples/ExactlyOnce)
in the GitHub repo.

<a id="consumer-dotnet-client"></a>

## Consumer

### Initialization

To create a .NET Consumer, first construct an instance of the `ConsumerConfig` class,
then pass this into the `ConsumerBuilder`’s constructor:

```bash
using System.Collections.Generic;
using Confluent.Kafka;

...

var config = new ConsumerConfig
{
    BootstrapServers = "host1:9092,host2:9092",
    GroupId = "foo",
    AutoOffsetReset = AutoOffsetReset.Earliest
};

using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
{
    ...
}
```

The `GroupId` property is mandatory and specifies which consumer group the
consumer is a member of. The `AutoOffsetReset` property specifies what
offset the consumer should start reading from ONLY in the event there are no
committed offsets for a partition, or the committed offset is invalid
(perhaps due to log truncation).

#### The Consume Loop

A typical Kafka consumer application is centered around a consume loop, which
repeatedly calls the `Consume` method to retrieve records one-by-one that
have been efficiently pre-fetched by the consumer in background threads. Before
entering the consume loop, you’ll typically use the `Subscribe` method to
join a group. The consumer instance will then be assigned to fetch from an
exclusive subset of the partitions managed by the group.

```bash
...

using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
{
    consumer.Subscribe(topics);

    while (!cancelled)
    {
        var consumeResult = consumer.Consume(cancellationToken);

        // handle consumed message.
        ...
    }

    consumer.Close();
}
```

Note that disposing the consumer instance after you are finished
using it (achieved with the `using` block in the above example)
will ensure that active sockets are closed and internal state is
cleaned up. In order to leave the group cleanly - i.e. commit
final offsets and trigger a group rebalance which ensures that
any partitions owned by the consumer are re-assigned to other
members in the group in a timely fashion - you additionally need
to call the `Close` method prior to disposing.

#### Auto Offset Commit

By default, the .NET Consumer will commit offsets automatically. This
is done periodically by a background thread at an interval specified
by the `AutoCommitIntervalMs` config property. An offset becomes
eligible to be committed immediately prior to being delivered to
the application via the `Consume` method.

This strategy introduces the potential for messages to be missed
in the case of application failure because the application may terminate
before it finishes processing a particular message, whilst the offset
corresponding to that message may be successfully committed to Kafka
by the background thread.

Furthermore, this strategy may also introduce duplicate
processing in the case of application failure since offsets are only
committed periodically.

#### Synchronous Commits

The C# client allows you to commit offsets explicitly via the
`Commit` method. In the following example, a synchronous commit
is triggered every `commitPeriod` messages:

```bash
var config = new ConsumerConfig
{
    ...
    // Disable auto-committing of offsets.
    EnableAutoCommit = false
}

...

while (!cancelled)
{
    var consumeResult = consumer.Consume(cancellationToken);

    // process message here.

    if (consumeResult.Offset % commitPeriod == 0)
    {
        try
        {
            consumer.Commit(consumeResult);
        }
        catch (KafkaException e)
        {
            Console.WriteLine($"Commit error: {e.Error.Reason}");
        }
    }
}
```

For processing failures that result in an application crash, the `Commit`
method provides “at least once” delivery semantics. In this scenario, the offset
corresponding to a message is only committed after the message has been
successfully processed. For errors where there is no application crash, the
application will call the [Seek()](https://docs.confluent.io/platform/current/clients/confluent-kafka-dotnet/_site/api/Confluent.Kafka.IConsumer-2.html#Confluent_Kafka_IConsumer_2_Seek_Confluent_Kafka_TopicPartitionOffset_)
method and move backward to the failed offset.

If you reverse the order of the processing and commit, as well as commit
before every message (not just periodically), you will get “at most once”
delivery semantics.

#### NOTE
You should generally avoid blocking network calls (including synchronous
use of `Commit`) because of the ramifications for throughput.

#### Store Offsets

The auto offset commit capability in the .NET Client is actually quite flexible.
As outlined above, by default, the offsets to be committed to Kafka are updated
immediately prior to the `Consume` method that delivers messages to the
application. However, you can prevent this from happening by setting the
`EnableAutoOffsetStore` config property to `false`. You can then use the
`StoreOffset` method to specify the offsets you would like to be committed by
the background thread. You can call this as many times as you like, but the
offset commit request will only sent to the group coordinator periodically with
the most recently specified offset(s) for the owned partitions. This approach is
preferred over the synchronous commit approach outlined in the previous section.

The below example uses this approach to achieve at least once delivery
semantics without blocking the main processing loop:

```bash
var config = new ConsumerConfig
{
    ...
    EnableAutoCommit = true // (the default)
    EnableAutoOffsetStore = false
}

...

while (!cancelled)
{
    var consumeResult = consumer.Consume(cancellationToken);

    // process message here.

    consumer.StoreOffset(consumeResult);
}
```

### Committing During a Rebalance

A consumer that is subscribed to a group is only permitted to commit offsets
for the subset of partitions that it owns. By default, a group rebalance
is handled automatically by the consumer and happens autonomously
behind the scenes. These semantics introduce a potential race condition
where you may try to commit or store an offset, but meanwhile a rebalance
has caused the consumer to lose ownership of the partition(s) you
are trying to commit the offset for. If this happens, the `Commit` or
`StoreOffset` call will throw a `KafkaException` with `ErrorCode`
equal to `Local_State` (“Erroneous state”). In the context of at-least
once processing, you can interpret this error as a warning that the
message will be processed again by another consumer in the group (the new
owner) and ignore it.

Alternatively, you can specify a (possibly empty) partitions revoked
handler. This will effectively avoid the race condition because rebalance
handlers are called as a side effect of a call to `Consume` on
the application thread - the rebalance is effectively blocked whilst
you process the consumed message and store or commit offsets.

```bash
...

using (var consumer = new ConsumerBuilder<Ignore, string>(config)
    .SetPartitionsRevokedHandler((c, partitions) =>
    {
        ...
    })
    .Build())
{
    ...
}
```

## API Documentation

Click [here](/platform/current/clients/confluent-kafka-dotnet/_site/api/Confluent.Kafka.html) to view the .NET Client API documentation.

## Related Content

- [Getting Started with Apache Kafka and .NET](https://developer.confluent.io/get-started/dotnet) (Confluent Developer)
- [Apache Kafka for .NET Developers](https://developer.confluent.io/learn-kafka/apache-kafka-for-dotnet/overview/) (Confluent Developer)
- [Designing the .NET API for Apache Kafka](https://www.confluent.io/blog/designing-the-net-api-for-apache-kafka) (Confluent Blog)
- [Implement a Cross-Platform Apache Kafka Producer and Consumer with C# and .NET](https://www.confluent.io/blog/build-cross-platform-kafka-applications-using-c-and-dotnet-5/) (Confluent Blog)