KafkaFlow

https://farfetch.github.io/kafkaflow/docs/
Admin
KafkaFlow is a .NET framework for building event-driven applications on top of Apache Kafka in a...
Tokens:95,168
Snippets:979
Trust Score:9.7
Update:2 hours ago
Show doc for...
Tokens:
Raw
### Configure KafkaFlow in Startup.cs

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/installation/

Configures KafkaFlow with a consumer and producer using dependency injection. Includes starting the Kafka bus on application startup. The example uses a sample topic, group ID, and JSON serialization.

```csharp
public void ConfigureServices(IServiceCollection services)
{
    services.AddKafka(kafka => kafka
        .UseConsoleLog()
        .AddCluster(cluster => cluster
            .WithBrokers(new[] { "localhost:9092" })
            .AddConsumer(consumer => consumer
                .Topic("sample-topic")
                .WithGroupId("sample-group")
                .WithBufferSize(100)
                .WithWorkersCount(10)
                .AddMiddlewares(middlewares => middlewares
                    .AddDeserializer<JsonCoreDeserializer>()
                    .AddTypedHandlers(handlers => handlers
                        .AddHandler<SampleMessageHandler>())
                )
            )
            .AddProducer("producer-name", producer => producer
                .DefaultTopic("sample-topic")
                .AddMiddlewares(middlewares => middlewares
                    .AddSerializer<JsonCoreSerializer>()
                )
            )
        )
    );
}

public void Configure(
    IApplicationBuilder app,
    IWebHostEnvironment env,
    IHostApplicationLifetime lifetime)
{
    var kafkaBus = app.ApplicationServices.CreateKafkaBus();

    lifetime.ApplicationStarted.Register(() => kafkaBus.StartAsync(lifetime.ApplicationStopped));
}
```

--------------------------------

### Configure KafkaFlow in ASP.NET Core Startup

Source: https://farfetch.github.io/kafkaflow/docs/guides/configuration/

Register KafkaFlow dependencies and start the Kafka bus before the application runs. This example uses the minimal hosting model with WebApplication.

```csharp
using KafkaFlow;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        ...
    )
);

var app = builder.Build();

app.MapGet("/", () => "Hello World!");

var kafkaBus = app.Services.CreateKafkaBus();
await kafkaBus.StartAsync();

app.Run();
```

--------------------------------

### Start consumer command

Source: https://farfetch.github.io/kafkaflow/docs/guides/admin/web-api/

C# command to start a consumer by name using IConsumerAdmin.

```csharp
var consumerAdmin = provider.GetService<IConsumerAdmin>();
await consumerAdmin.StartConsumerAsync(consumerName);
```

--------------------------------

### Start the Kafka cluster with Docker Compose

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/create-your-first-application/

Run this command in the terminal to start the Apache Kafka cluster defined in the docker-compose.yml file.

```bash
docker-compose up -d
```

--------------------------------

### Install KafkaFlow packages for Producer

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/create-your-first-application/

Run these commands inside the Producer project directory to install the required KafkaFlow and dependency injection packages.

```bash
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package KafkaFlow.LogHandler.Console
dotnet add package KafkaFlow.Serializer.JsonCore
dotnet add package Microsoft.Extensions.DependencyInjection
```

--------------------------------

### Install KafkaFlow Admin packages

Source: https://farfetch.github.io/kafkaflow/docs/guides/admin/web-api/

Install the required NuGet packages for the Administration Web API: KafkaFlow.Admin and KafkaFlow.Admin.WebApi. Also includes the base KafkaFlow.Microsoft.DependencyInjection package.

```bash
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package KafkaFlow.Admin
dotnet add package KafkaFlow.Admin.WebApi
```

--------------------------------

### Install Swashbuckle package

Source: https://farfetch.github.io/kafkaflow/docs/guides/admin/web-api/

Command to install the Swashbuckle.AspNetCore package for Swagger support.

```bash
dotnet add package Swashbuckle.AspNetCore
```

--------------------------------

### StartConsumerByName.ConsumerName property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow.Admin/KafkaFlow.Admin.Messages/StartConsumerByName/ConsumerName/

Gets or sets the consumer name that will be started.

```APIDOC
## StartConsumerByName.ConsumerName property

### Description
Gets or sets the consumer name that will be started.

### Property Signature
```csharp
public string ConsumerName { get; set; }
```

### Property Type
- **string** - Gets or sets the consumer name.

### See Also
- class StartConsumerByName
- namespace KafkaFlow.Admin.Messages
```

--------------------------------

### Install KafkaFlow Dashboard packages

Source: https://farfetch.github.io/kafkaflow/docs/guides/admin/dashboard/

Install the required packages for the Dashboard: KafkaFlow.Admin, KafkaFlow.Admin.WebApi, and KafkaFlow.Admin.Dashboard. Also includes the base KafkaFlow.Microsoft.DependencyInjection package.

```bash
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package KafkaFlow.Admin
dotnet add package KafkaFlow.Admin.WebApi
dotnet add package KafkaFlow.Admin.Dashboard
```

--------------------------------

### ClusterConfiguration.OnStartedHandler property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/ClusterConfiguration/OnStartedHandler/

Gets the handler to be executed when the cluster started. This property is read-only and returns an Action<IDependencyResolver> delegate.

```APIDOC
## ClusterConfiguration.OnStartedHandler property

### Description
Gets the handler to be executed when the cluster started.

### Property Signature
```csharp
public Action<IDependencyResolver> OnStartedHandler { get; }
```

### Property Value
- **Type**: `Action<IDependencyResolver>`
- **Access**: Read-only (getter only)

### Remarks
This property is part of the `ClusterConfiguration` class in the `KafkaFlow.Configuration` namespace.
```

--------------------------------

### IConsumerConfiguration.Topics

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IConsumerConfiguration/Topics/

Gets the consumer configured topics.

```APIDOC
## IConsumerConfiguration.Topics

### Description
Gets the consumer configured topics.

### Property
```csharp
public IReadOnlyList<string> Topics { get; }
```

### Returns
- **IReadOnlyList<string>** - The list of configured topics.

### See Also
- interface IConsumerConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### Configure KafkaFlow Dashboard in .NET 6

Source: https://farfetch.github.io/kafkaflow/docs/guides/admin/dashboard/

Configures the Dashboard during application setup. Requires installing the packages listed above. The Dashboard UI is available at /kafkaflow and refreshes every 5 seconds with telemetry data at /kafkaflow/consumers/telemetry.

```csharp
using KafkaFlow;
using KafkaFlow.Admin.Dashboard;

var builder = WebApplication.CreateBuilder(args);

builder.Services
    .AddKafka(kafka => kafka
        .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddConsumer(consumer => consumer
                    .Topic("mytopic")
                    .WithGroupId("g1")
                    .WithWorkersCount(1)
                    .WithBufferSize(10)
                )
                .EnableAdminMessages("kafka-flow.admin")
                .EnableTelemetry("kafka-flow.admin") // you can use the same topic used in EnableAdminMessages, if need it
        ))
    .AddControllers();

var app = builder.Build();

app.MapControllers();
app.UseKafkaFlowDashboard();

var kafkaBus = app.Services.CreateKafkaBus();
await kafkaBus.StartAsync();

await app.RunAsync();
```

--------------------------------

### IConsumerAccessor.All property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/IConsumerAccessor/All/

Gets all configured consumers.

```APIDOC
## IConsumerAccessor.All property

### Description
Gets all configured consumers.

### Property Signature
```csharp
public IEnumerable<IMessageConsumer> All { get; }
```

### Property Value
- **Type**: IEnumerable<IMessageConsumer>
- **Description**: An enumerable of all configured consumers.

### See Also
- interface IMessageConsumer
- interface IConsumerAccessor
- namespace KafkaFlow.Consumers
```

--------------------------------

### IKafkaBus.StartAsync(CancellationToken)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow/IKafkaBus/StartAsync/

Starts all consumers. Accepts an optional CancellationToken to stop the operation.

```APIDOC
## StartAsync(CancellationToken)

### Description
Starts all consumers.

### Method
StartAsync

### Signature
public Task StartAsync(CancellationToken stopCancellationToken = default)

### Parameters
- **stopCancellationToken** (CancellationToken) - Optional - A CancellationToken used to stop the operation.

### Returns
Task
```

--------------------------------

### StartConsumerAsync(string consumerName)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow.Admin/KafkaFlow.Admin/IConsumerAdmin/StartConsumerAsync/

Starts a consumer with the specified unique name.

```APIDOC
## StartConsumerAsync(string consumerName)

### Description
Starts a consumer with the specified unique name.

### Method
StartConsumerAsync

### Parameters
#### Parameters
- **consumerName** (string) - Required - The consumer unique name.

### Return Value
- **Task** - A task representing the asynchronous operation.

```

--------------------------------

### IMessageConsumer.StartAsync()

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/IMessageConsumer/StartAsync/

Starts KafkaFlow consumer creating a new Confluent Consumer.

```APIDOC
## IMessageConsumer.StartAsync()

### Description
Starts KafkaFlow consumer creating a new Confluent Consumer.

### Method
StartAsync

### Signature
```csharp
public Task StartAsync()
```

### Parameters
None

### Returns
- **Task** - A task representing the asynchronous operation.

### See Also
- interface IMessageConsumer
- namespace KafkaFlow.Consumers
```

--------------------------------

### IMessageConsumer.RunningPartitions property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/IMessageConsumer/RunningPartitions/

Gets the consumer's running partitions.

```APIDOC
## IMessageConsumer.RunningPartitions property

### Description
Gets the consumer's running partitions.

### Property Signature
```csharp
public IEnumerable<TopicPartition> RunningPartitions { get; }
```

### Return Value
- **IEnumerable<TopicPartition>** - The collection of running partitions.

### See Also
- interface IMessageConsumer
- namespace KafkaFlow.Consumers
```

--------------------------------

### IConsumerConfiguration.PartitionsAssignedHandlers property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IConsumerConfiguration/PartitionsAssignedHandlers/

Gets the handlers that will be called when the partitions are assigned.

```APIDOC
## IConsumerConfiguration.PartitionsAssignedHandlers property

### Description
Gets the handlers that will be called when the partitions are assigned.

### Property Signature
```csharp
public IReadOnlyList<Action<IDependencyResolver, List<TopicPartition>>> PartitionsAssignedHandlers { get; }
```

### Property Value
- **Type**: IReadOnlyList<Action<IDependencyResolver, List<TopicPartition>>>
- **Description**: The list of handlers to be called when partitions are assigned.

### See Also
- interface IConsumerConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### ResumeConsumerByName.Topics property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow.Admin/KafkaFlow.Admin.Messages/ResumeConsumerByName/Topics/

Gets or sets the topics that will be resumed.

```APIDOC
## ResumeConsumerByName.Topics property

### Description
Gets or sets the topics that will be resumed.

### Property Signature
```csharp
public IList<string> Topics { get; set; }
```

### Property Value
- **Type**: IList<string>
- **Access**: Get/Set

### See Also
- class ResumeConsumerByName
- namespace KafkaFlow.Admin.Messages
```

--------------------------------

### IProducerConfiguration.MiddlewaresConfigurations property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IProducerConfiguration/MiddlewaresConfigurations/

Gets the middlewares configurations for the producer configuration.

```APIDOC
## IProducerConfiguration.MiddlewaresConfigurations property

### Description
Gets the middlewares configurations.

### Property Signature
```csharp
public IReadOnlyList<MiddlewareConfiguration> MiddlewaresConfigurations { get; }
```

### Property Value
- **Type**: IReadOnlyList<MiddlewareConfiguration>
- **Description**: The list of middleware configurations.

### See Also
- class MiddlewareConfiguration
- interface IProducerConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### Add required packages for ASP.NET Core Startup

Source: https://farfetch.github.io/kafkaflow/docs/guides/configuration/

Install the required NuGet packages for configuring KafkaFlow in an ASP.NET Core application.

```bash
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Microsoft.DependencyInjection
```

--------------------------------

### Create the Consumer console project

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/create-your-first-application/

Use the .NET CLI to create a new console project named Consumer.

```bash
dotnet new console --name Consumer
```

--------------------------------

### Run the Consumer

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/create-your-first-application/

Command to run the consumer project from the KafkaFlowQuickstart directory.

```bash
dotnet run --project Consumer/Consumer.csproj 
```

--------------------------------

### MiddlewareConfiguration.InstanceContainerId property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/MiddlewareConfiguration/InstanceContainerId/

Gets the instance container ID used to get the desired factory when creating the instance.

```APIDOC
## MiddlewareConfiguration.InstanceContainerId property

### Description
Gets the instance container ID used to get the desired factory when creating the instance.

### Property Signature
```csharp
public Guid? InstanceContainerId { get; }
```

### Property Value
- **Type**: `Guid?` (nullable Guid)
- **Access**: Read-only (getter only)

### Remarks
This property is part of the `MiddlewareConfiguration` class in the `KafkaFlow.Configuration` namespace.
```

--------------------------------

### Create the Producer console project

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/create-your-first-application/

Use the .NET CLI to create a new console project named Producer.

```bash
dotnet new console --name Producer
```

--------------------------------

### Configure and send a message with KafkaFlow producer

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/create-your-first-application/

Replace Program.cs with this code to set up the KafkaFlow producer, configure the broker, create the topic, and send a HelloMessage.

```csharp
using Microsoft.Extensions.DependencyInjection;
using KafkaFlow.Producers;
using KafkaFlow.Serializer;
using KafkaFlow;
using Producer;

var services = new ServiceCollection();

const string topicName = "sample-topic";
const string producerName = "say-hello";

services.AddKafka(
    kafka => kafka
        .UseConsoleLog()
        .AddCluster(
            cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .CreateTopicIfNotExists(topicName, 1, 1)
                .AddProducer(
                    producerName,
                    producer => producer
                        .DefaultTopic(topicName)
                        .AddMiddlewares(m =>
                            m.AddSerializer<JsonCoreSerializer>()
                            )
                )
        )
);

var serviceProvider = services.BuildServiceProvider();

var producer = serviceProvider
    .GetRequiredService<IProducerAccessor>()
    .GetProducer(producerName);

await producer.ProduceAsync(
                   topicName,
                   Guid.NewGuid().ToString(),
                   new HelloMessage { Text = "Hello!" });


Console.WriteLine("Message sent!");
```

--------------------------------

### PartitionKeyDistributionStrategy.Initialize(IReadOnlyList<IWorker>)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers.DistributionStrategies/PartitionKeyDistributionStrategy/Initialize/

Initializes the distribution strategy with the provided list of workers.

```APIDOC
## PartitionKeyDistributionStrategy.Initialize(IReadOnlyList<IWorker>)

### Description
Initializes the distribution strategy with the provided list of workers.

### Method
Initialize

### Signature
public void Initialize(IReadOnlyList<IWorker> workers)

### Parameters
- **workers** (IReadOnlyList<IWorker>) - The list of workers to initialize the strategy with.

### Return Type
void

### See Also
- class PartitionKeyDistributionStrategy
- namespace KafkaFlow.Consumers.DistributionStrategies
```

--------------------------------

### PartitionKeyDistributionStrategy()

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers.DistributionStrategies/PartitionKeyDistributionStrategy/

The default constructor for the PartitionKeyDistributionStrategy class. It initializes a new instance of the strategy.

```APIDOC
## PartitionKeyDistributionStrategy()

### Description
Initializes a new instance of the PartitionKeyDistributionStrategy class.

### Method
Constructor

### Signature
```csharp
public PartitionKeyDistributionStrategy()
```

### Parameters
None.

### Returns
A new instance of the PartitionKeyDistributionStrategy class.

### See Also
- class PartitionKeyDistributionStrategy
- namespace KafkaFlow.Consumers.DistributionStrategies
```

--------------------------------

### IProducerAccessor Interface

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Producers/IProducerAccessor/

The IProducerAccessor interface provides access to the configured producers. It includes properties to get all producers or a producer by name, and methods to get a producer by name or by type.

```APIDOC
## IProducerAccessor Interface

### Description
Provides access to the configured producers.

### Namespace
KafkaFlow.Producers

### Definition
```csharp
public interface IProducerAccessor
```

### Members

#### All { get; }
Gets all configured producers.

#### Item { get; }
Gets a producer by its name.

#### GetProducer(...)
Gets a producer by its name.

#### GetProducer<TProducer>()
Gets a producer by its type.

### See Also
- namespace KafkaFlow.Producers
```

--------------------------------

### Create the Message Consumer (Program.cs)

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/create-your-first-application/

Replaces the content of Program.cs with a KafkaFlow consumer that subscribes to 'sample-topic', uses the 'sample-group' consumer group, and handles messages with HelloMessageHandler. It configures a buffer size of 100, 10 workers, and JSON deserialization.

```csharp
using KafkaFlow;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;
using Consumer;

const string topicName = "sample-topic";
var services = new ServiceCollection();

services.AddKafka(kafka => kafka
    .UseConsoleLog()
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .CreateTopicIfNotExists(topicName, 1, 1)
        .AddConsumer(consumer => consumer
            .Topic(topicName)
            .WithGroupId("sample-group")
            .WithBufferSize(100)
            .WithWorkersCount(10)
            .AddMiddlewares(middlewares => middlewares
                .AddDeserializer<JsonCoreDeserializer>()
                .AddTypedHandlers(h => h.AddHandler<HelloMessageHandler>())
            )
        )
    )
);

var serviceProvider = services.BuildServiceProvider();

var bus = serviceProvider.CreateKafkaBus();

await bus.StartAsync();

Console.ReadKey();

await bus.StopAsync();
```

--------------------------------

### IProducerAccessor indexer

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Producers/IProducerAccessor/Item/

Gets a producer by its name.

```APIDOC
## IProducerAccessor indexer

### Description
Gets a producer by its name.

### Method
Indexer

### Endpoint
IProducerAccessor.this[string name]

### Parameters
#### Parameters
- **name** (string) - Required - producer name

### Response
#### Success Response
- **IMessageProducer** - The producer associated with the given name.

### See Also
- interface IMessageProducer
- interface IProducerAccessor
- namespace KafkaFlow.Producers
```

--------------------------------

### MessageHeaders.GetEnumerator()

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow/MessageHeaders/GetEnumerator/

Gets an enumerator that iterates through Headers.

```APIDOC
## MessageHeaders.GetEnumerator()

### Description
Gets an enumerator that iterates through Headers.

### Method
GetEnumerator

### Signature
public IEnumerator<KeyValuePair<string, byte[]>> GetEnumerator()

### Return Type
IEnumerator<KeyValuePair<string, byte[]>>

### See Also
- class MessageHeaders
- namespace KafkaFlow
```

--------------------------------

### BatchProduceItem(string topic, object messageKey, object messageValue, IMessageHeaders headers, int partition)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Producers/BatchProduceItem/

Initializes a new instance of the BatchProduceItem class with the specified topic, message key, message value, headers, and partition.

```APIDOC
## BatchProduceItem Constructor (2 of 2)

### Description
Initializes a new instance of the `BatchProduceItem` class.

### Method
Constructor

### Signature
```csharp
public BatchProduceItem(string topic, object messageKey, object messageValue, IMessageHeaders headers, int partition)
```

### Parameters
- **topic** (string) - Required - The destination topic
- **messageKey** (object) - Required - The message partition key
- **messageValue** (object) - Required - The message content
- **headers** (IMessageHeaders) - Required - The message headers
- **partition** (int) - Required - The partition to produce the message to

### See Also
- class `BatchProduceItem`
- namespace `KafkaFlow.Producers`
```

--------------------------------

### IMessageConsumer.Topics

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/IMessageConsumer/Topics/

Gets the consumer configured topics.

```APIDOC
## IMessageConsumer.Topics

### Description
Gets the consumer configured topics.

### Property Signature
```csharp
public IReadOnlyList<string> Topics { get; }
```

### Property Value
- **Type**: IReadOnlyList<string>
- **Description**: The list of configured topics.

### Remarks
This property is part of the IMessageConsumer interface in the KafkaFlow.Consumers namespace.
```

--------------------------------

### IConsumer.FlowManager property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/IConsumer/FlowManager/

Gets the consumer IConsumerFlowManager.

```APIDOC
## IConsumer.FlowManager property

### Description
Gets the consumer `IConsumerFlowManager`.

### Property Signature
```csharp
public IConsumerFlowManager FlowManager { get; }
```

### See Also
- interface IConsumerFlowManager
- interface IConsumer
- namespace KafkaFlow.Consumers
```

--------------------------------

### IProducerConfiguration.Cluster

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IProducerConfiguration/Cluster/

Gets the cluster configuration for the producer.

```APIDOC
## IProducerConfiguration.Cluster

### Description
Gets the cluster configuration for the producer.

### Property Signature
```csharp
public ClusterConfiguration Cluster { get; }
```

### Property Value
- **ClusterConfiguration** - The cluster configuration.

### See Also
- class ClusterConfiguration
- interface IProducerConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### Implement the HelloMessageHandler

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/create-your-first-application/

Add this class to the Consumer project to handle incoming HelloMessage instances and write the partition, offset, and message text to the console.

```csharp
using KafkaFlow;
using Producer;

namespace Consumer;

public class HelloMessageHandler : IMessageHandler<HelloMessage>
{
    public Task Handle(IMessageContext context, HelloMessage message)
    {
        Console.WriteLine(
            "Partition: {0} | Offset: {1} | Message: {2}",
            context.ConsumerContext.Partition,
            context.ConsumerContext.Offset,
            message.Text);

        return Task.CompletedTask;
    }
}
```

--------------------------------

### StartConsumerByName()

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow.Admin/KafkaFlow.Admin.Messages/StartConsumerByName/

The default constructor for the StartConsumerByName class. It initializes a new instance of the StartConsumerByName message.

```APIDOC
## StartConsumerByName()

### Description
Initializes a new instance of the StartConsumerByName class.

### Method
Constructor

### Signature
```csharp
public StartConsumerByName()
```

### Parameters
None.

### Returns
A new instance of the StartConsumerByName class.

### See Also
- class StartConsumerByName
- namespace KafkaFlow.Admin.Messages
```

--------------------------------

### ClusterConfiguration.Brokers property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/ClusterConfiguration/Brokers/

Gets the list of brokers.

```APIDOC
## ClusterConfiguration.Brokers property

### Description
Gets the list of brokers.

### Property Signature
```csharp
public IReadOnlyCollection<string> Brokers { get; }
```

### Property Value
- **Type**: IReadOnlyCollection<string>
- **Description**: The list of brokers.

### See Also
- class ClusterConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### BatchProduceItem(string topic, object messageKey, object messageValue, IMessageHeaders headers)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Producers/BatchProduceItem/

Initializes a new instance of the BatchProduceItem class with the specified topic, message key, message value, and headers.

```APIDOC
## BatchProduceItem Constructor (1 of 2)

### Description
Initializes a new instance of the `BatchProduceItem` class.

### Method
Constructor

### Signature
```csharp
public BatchProduceItem(string topic, object messageKey, object messageValue, IMessageHeaders headers)
```

### Parameters
- **topic** (string) - Required - The destination topic
- **messageKey** (object) - Required - The message partition key
- **messageValue** (object) - Required - The message content
- **headers** (IMessageHeaders) - Required - The message headers

### See Also
- class `BatchProduceItem`
- namespace `KafkaFlow.Producers`
```

--------------------------------

### CompressorProducerMiddleware(ICompressor compressor)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Middlewares.Compressor/CompressorProducerMiddleware/

Initializes a new instance of the CompressorProducerMiddleware class with the specified compressor.

```APIDOC
## CompressorProducerMiddleware(ICompressor compressor)

### Description
Initializes a new instance of the `CompressorProducerMiddleware` class.

### Method
Constructor

### Signature
```csharp
public CompressorProducerMiddleware(ICompressor compressor)
```

### Parameters
- **compressor** (ICompressor) - Required - Instance of ICompressor

### See Also
- class CompressorProducerMiddleware
- namespace KafkaFlow.Middlewares.Compressor
```

--------------------------------

### ClusterConfiguration.Kafka property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/ClusterConfiguration/Kafka/

Gets the kafka configuration.

```APIDOC
## ClusterConfiguration.Kafka property

### Description
Gets the kafka configuration.

### Property Signature
```csharp
public KafkaConfiguration Kafka { get; }
```

### Property Value
- **KafkaConfiguration** - The kafka configuration.

### See Also
- class KafkaConfiguration
- class ClusterConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### Install KafkaFlow packages using .NET CLI

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/installation/

Commands to add the required KafkaFlow NuGet packages: KafkaFlow, KafkaFlow.Microsoft.DependencyInjection, and KafkaFlow.LogHandler.Console.

```shell
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package KafkaFlow.LogHandler.Console
```

--------------------------------

### IClusterManager interface

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Clusters/IClusterManager/

Provides access to Cluster administration. This interface includes properties and methods for managing clusters, such as getting the cluster name, creating topics if they don't exist, retrieving consumer group offsets, and getting topic metadata.

```APIDOC
## IClusterManager interface

### Description
Provides access to Cluster administration.

### Interface Definition
```csharp
public interface IClusterManager
```

### Members

#### ClusterName { get; }
Gets the unique cluster name defined in the configuration.

#### CreateIfNotExistsAsync(...)
Create Topics when building Cluster Configuration if the topic doesn't exist.

#### GetConsumerGroupOffsetsAsync(...)
Retrieves the offsets for a specified consumer group and a collection of topics. The offsets indicate where the consumer group is in its consumption of the topics.

#### GetTopicMetadataAsync(...)
Retrieves the metadata for a specified topic. This includes information about the topic as well as its partitions.

### See Also
- namespace KafkaFlow.Clusters
```

--------------------------------

### MessageHeaders.GetKafkaHeaders()

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow/MessageHeaders/GetKafkaHeaders/

Gets all the kafka headers from the MessageHeaders instance.

```APIDOC
## MessageHeaders.GetKafkaHeaders()

### Description
Gets all the kafka headers.

### Method
GetKafkaHeaders

### Endpoint
MessageHeaders.GetKafkaHeaders()

### Parameters
None

### Request Example
```csharp
public Headers GetKafkaHeaders()
```

### Response
#### Success Response
- **Headers** - The kafka headers.

#### Response Example
```csharp
Headers headers = messageHeaders.GetKafkaHeaders();
```
```

--------------------------------

### PendingOffsetsStatisticsHandler.Interval property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/PendingOffsetsStatisticsHandler/Interval/

Gets the interval at which the handler should be executed.

```APIDOC
## PendingOffsetsStatisticsHandler.Interval property

### Description
Gets the interval at which the handler should be executed.

### Property Signature
```csharp
public TimeSpan Interval { get; }
```

### See Also
- class PendingOffsetsStatisticsHandler
- namespace KafkaFlow.Configuration
```

--------------------------------

### TopicPartitionMetadata constructor

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow/TopicPartitionMetadata/

Initializes a new instance of the TopicPartitionMetadata class with the specified partition id.

```APIDOC
## TopicPartitionMetadata constructor

### Description
Initializes a new instance of the TopicPartitionMetadata class with the specified partition id.

### Method
Constructor

### Signature
```csharp
public TopicPartitionMetadata(int id)
```

### Parameters
- **id** (int) - Required - The partition id.

### See Also
- class TopicPartitionMetadata
- namespace KafkaFlow
```

--------------------------------

### KafkaConfiguration.Clusters property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/KafkaConfiguration/Clusters/

Gets the cluster configuration list.

```APIDOC
## KafkaConfiguration.Clusters property

### Description
Gets the cluster configuration list.

### Property Signature
```csharp
public IReadOnlyCollection<ClusterConfiguration> Clusters { get; }
```

### Property Value
- **Type**: IReadOnlyCollection<ClusterConfiguration>
- **Description**: The cluster configuration list.

### See Also
- class ClusterConfiguration
- class KafkaConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### MiddlewareConfiguration.Lifetime property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/MiddlewareConfiguration/Lifetime/

Gets the middleware instance lifetime.

```APIDOC
## MiddlewareConfiguration.Lifetime property

### Description
Gets the middleware instance lifetime.

### Property Signature
```csharp
public MiddlewareLifetime Lifetime { get; }
```

### Property Value
- **Type**: MiddlewareLifetime
- **Access**: Read-only

### See Also
- class MiddlewareConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### Add required packages for Hosted Service

Source: https://farfetch.github.io/kafkaflow/docs/guides/configuration/

Install the necessary NuGet packages to use KafkaFlow with the Hosted Service model in a console application.

```bash
dotnet add package KafkaFlow
dotnet add package KafkaFlow.Extensions.Hosting
dotnet add package KafkaFlow.Microsoft.DependencyInjection
dotnet add package Microsoft.Extensions.Hosting
```

--------------------------------

### IProducerConfiguration.BaseProducerConfig property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IProducerConfiguration/BaseProducerConfig/

Gets the confluent producer configuration.

```APIDOC
## IProducerConfiguration.BaseProducerConfig property

### Description
Gets the confluent producer configuration.

### Property Signature
```csharp
public ProducerConfig BaseProducerConfig { get; }
```

### See Also
- interface IProducerConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### IConsumerConfiguration.ClusterConfiguration property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IConsumerConfiguration/ClusterConfiguration/

Gets the cluster configuration for the consumer.

```APIDOC
## IConsumerConfiguration.ClusterConfiguration property

### Description
Gets the cluster configuration.

### Property Signature
```csharp
public ClusterConfiguration ClusterConfiguration { get; }
```

### Property Type
- **ClusterConfiguration** - The cluster configuration.

### See Also
- class ClusterConfiguration
- interface IConsumerConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### AddCompressor with GzipMessageCompressor in Startup

Source: https://farfetch.github.io/kafkaflow/docs/guides/middlewares/compressor-middleware/

Shows how to add the GzipMessageCompressor to a producer's middlewares in the Startup class. The AddCompressor method accepts a class implementing ICompressor as a generic argument, or a factory method can provide an instance. Requires the KafkaFlow.Compressor.Gzip package.

```csharp
public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddKafka(kafka => kafka
            .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddProducer<ProductEventsProducer>(producer => producer
                    ...
                    .AddMiddlewares(middlewares => middlewares
                        ...
                        .AddCompressor<GzipMessageCompressor>()
                        // or
                        .AddCompressor(resolver => new GzipMessageCompressor(...))
                        ...
                    )
                )
            )
        );
    }
}
```

--------------------------------

### Configure Consumer Throttling with ByOtherConsumersLag

Source: https://farfetch.github.io/kafkaflow/docs/guides/middlewares/consumer-throttling-middleware/

Shows how to enable throttling on a consumer by monitoring another consumer's lag and applying delays at different thresholds. The example uses the fluent API with ThrottleConsumer, ByOtherConsumersLag, WithInterval, and AddAction methods.

```csharp
.AddConsumer(
    consumer => consumer
        .Topic("bulk-topic")
        .WithName("bulkConsumer")
        .AddMiddlewares(
            middlewares => middlewares
                .ThrottleConsumer(
                    t => t
                        .ByOtherConsumersLag("singleConsumer")
                        .WithInterval(TimeSpan.FromSeconds(5))
                        .AddAction(a => a.AboveThreshold(10).ApplyDelay(100))
                        .AddAction(a => a.AboveThreshold(100).ApplyDelay(1_000))
                        .AddAction(a => a.AboveThreshold(1_000).ApplyDelay(10_000)))
                .AddDeserializer<JsonCoreDeserializer>()
        )
)
```

--------------------------------

### IProducerConfiguration.CustomFactory property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IProducerConfiguration/CustomFactory/

Gets the producer custom factory.

```APIDOC
## IProducerConfiguration.CustomFactory property

### Description
Gets the producer custom factory.

### Property Signature
```csharp
public ProducerCustomFactory CustomFactory { get; }
```

### Property Value
- **Type**: `ProducerCustomFactory` (delegate)
- **Access**: Read-only (getter only)

### See Also
- delegate `ProducerCustomFactory`
- interface `IProducerConfiguration`
- namespace `KafkaFlow.Configuration`
```

--------------------------------

### TopicMetadata(string name, IReadOnlyCollection<TopicPartitionMetadata> partitions)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow/TopicMetadata/

Initializes a new instance of the TopicMetadata class with the specified topic name and partition metadata collection.

```APIDOC
## TopicMetadata(string name, IReadOnlyCollection<TopicPartitionMetadata> partitions)

### Description
Initializes a new instance of the TopicMetadata class with the specified topic name and partition metadata collection.

### Method
Constructor

### Signature
```csharp
public TopicMetadata(string name, IReadOnlyCollection<TopicPartitionMetadata> partitions)
```

### Parameters
- **name** (string) - Required - The name of the topic.
- **partitions** (IReadOnlyCollection<TopicPartitionMetadata>) - Required - A collection of partition metadata for the topic.

### See Also
- class TopicPartitionMetadata
- record TopicMetadata
- namespace KafkaFlow
```

--------------------------------

### GetProducer<TProducer>()

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Producers/IProducerAccessor/GetProducer/

Gets a producer by its type as defined in the configuration.

```APIDOC
## GetProducer<TProducer>()

### Description
Gets a producer by its type.

### Method
GetProducer<TProducer>

### Type Parameters
- **TProducer** - The type defined in the configuration.

### Returns
- **IMessageProducer** - The producer instance.

### See Also
- interface IMessageProducer
- interface IProducerAccessor
- namespace KafkaFlow.Producers
```

--------------------------------

### Handle No Handler Found Event with WhenNoHandlerFound

Source: https://farfetch.github.io/kafkaflow/docs/guides/middlewares/typed-handler-middleware/

Shows how to handle the event when no handler is found for an arriving message. The example writes a message to the console with the partition and offset from the consumer context. This is useful for logging or monitoring unhandled messages.

```csharp
services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddConsumer(consumer => consumer
            ...
            .AddMiddlewares(middlewares => middlewares
                ...
                .AddTypedHandlers(handlers => handlers
                    .AddHandler<ProductCreatedHandler>()
                    .WhenNoHandlerFound(context => 
                        Console.WriteLine("Message not handled > Partition: {0} | Offset: {1}",
                            context.ConsumerContext.Partition,
                            context.ConsumerContext.Offset)
                        )
                )
            )
        )
    )
);
```

--------------------------------

### Consumer configuration before v3 with AddCompressor and AddSerializer

Source: https://farfetch.github.io/kafkaflow/docs/guides/migration/from-v2-to-v3/

Shows the old consumer configuration using AddCompressor and AddSerializer. This is the pre-v3 pattern that must be updated.

```csharp
.AddConsumer(
    consumerBuilder => consumerBuilder
        .Topic("test-topic")
        .WithGroupId("group1")
        .AddMiddlewares(
            middlewares => middlewares
                .AddCompressor<GzipMessageCompressor>()
                .AddSerializer<JsonCoreSerializer>()
        )
)
```

--------------------------------

### TopicPartitionLag.Topic property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/TopicPartitionLag/Topic/

Gets the topic name for a TopicPartitionLag instance.

```APIDOC
## TopicPartitionLag.Topic property

### Description
Gets the topic name.

### Property Signature
```csharp
public string Topic { get; }
```

### Property Value
- **Type**: string
- **Access**: Read-only (getter only)

### See Also
- class TopicPartitionLag
- namespace KafkaFlow.Consumers
```

--------------------------------

### Import KafkaFlow namespaces

Source: https://farfetch.github.io/kafkaflow/docs/getting-started/installation/

Using directives for the KafkaFlow, KafkaFlow.Producers, and KafkaFlow.Serializer namespaces, required to access the types used in the setup.

```csharp
using KafkaFlow;
using KafkaFlow.Producers;
using KafkaFlow.Serializer;
```

--------------------------------

### TopicPartitionLag.Partition property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/TopicPartitionLag/Partition/

Gets the partition value for a TopicPartitionLag instance.

```APIDOC
## TopicPartitionLag.Partition property

### Description
Gets the partition value.

### Property Signature
```csharp
public int Partition { get; }
```

### Property Value
- **Type**: int
- **Access**: Read-only (getter only)

### See Also
- class TopicPartitionLag
- namespace KafkaFlow.Consumers
```

--------------------------------

### TopicConfiguration.Configs property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/TopicConfiguration/Configs/

Gets the topic creation configuration as a dictionary of string keys and values.

```APIDOC
## TopicConfiguration.Configs property

### Description
Gets the topic creation configuration.

### Property Signature
```csharp
public Dictionary<string, string> Configs { get; }
```

### Property Value
- **Type**: `Dictionary<string, string>`
- **Description**: The topic creation configuration.

### See Also
- class `TopicConfiguration`
- namespace `KafkaFlow.Configuration`
```

--------------------------------

### TopicConfiguration Constructor

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/TopicConfiguration/

Initializes a new instance of the TopicConfiguration class with the specified topic name, partition count, replication factor, and additional configuration values.

```APIDOC
## TopicConfiguration Constructor

### Description
Initializes a new instance of the `TopicConfiguration` class.

### Method
Constructor

### Endpoint
`TopicConfiguration(string name, int partitions, short replicas, Dictionary<string, string> configs)`

### Parameters
#### Parameters
- **name** (string) - Required - The topic name
- **partitions** (int) - Required - The number of partitions for the topic
- **replicas** (short) - Required - Replication factor for the topic
- **configs** (Dictionary<string, string>) - Required - Additional topic creation configuration values.

### Request Example
```csharp
public TopicConfiguration(string name, int partitions, short replicas, 
    Dictionary<string, string> configs)
```

### Response
#### Success Response
No return value (void).

#### Response Example
No response body.

### See Also
- class `TopicConfiguration`
- namespace `KafkaFlow.Configuration`
```

--------------------------------

### IConsumerFlowManager.PausedPartitions property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/IConsumerFlowManager/PausedPartitions/

Gets a list of the consumer paused partitions.

```APIDOC
## IConsumerFlowManager.PausedPartitions property

### Description
Gets a list of the consumer paused partitions.

### Property Signature
```csharp
public IReadOnlyList<TopicPartition> PausedPartitions { get; }
```

### Property Type
- **IReadOnlyList<TopicPartition>** - A read-only list of TopicPartition objects representing the paused partitions.

### Remarks
- This property is part of the IConsumerFlowManager interface in the KafkaFlow.Consumers namespace.

```

--------------------------------

### IConsumer.Configuration property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Consumers/IConsumer/Configuration/

Gets the consumer configuration for an IConsumer instance.

```APIDOC
## IConsumer.Configuration property

### Description
Gets the consumer configuration.

### Property Signature
```csharp
public IConsumerConfiguration Configuration { get; }
```

### Returns
- **IConsumerConfiguration** - The consumer configuration.

### See Also
- interface IConsumerConfiguration
- interface IConsumer
- namespace KafkaFlow.Consumers
```

--------------------------------

### Full Producer Configuration with Compression

Source: https://farfetch.github.io/kafkaflow/docs/guides/compression/

Shows a complete KafkaFlow producer setup with compression enabled. The configuration must be done only by producers; consumers automatically decompress messages.

```csharp
services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddProducer<ProductEventsProducer>(producer => producer
            .WithCompression(CompressionType.Gzip)
                ...
            )
        )
    )
);
```

--------------------------------

### Consumer configuration after v3 with AddDecompressor and AddDeserializer

Source: https://farfetch.github.io/kafkaflow/docs/guides/migration/from-v2-to-v3/

Shows the updated consumer configuration using AddDecompressor and AddDeserializer. This is the v3 pattern that replaces the old one.

```csharp
.AddConsumer(
    consumerBuilder => consumerBuilder
        .Topic("test-topic")
        .WithGroupId("group1")
        .AddMiddlewares(
            middlewares => middlewares
                .AddDecompressor<GzipMessageDecompressor>()
                .AddDeserializer<JsonCoreDeserializer>()
        )
)
```

--------------------------------

### IProducerConfiguration.Acks property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IProducerConfiguration/Acks/

Gets the acknowledge type for the producer configuration.

```APIDOC
## IProducerConfiguration.Acks property

### Description
Gets the acknowledge type for the producer configuration.

### Property Signature
```csharp
public Acks? Acks { get; }
```

### Property Value
- **Acks?** - The acknowledge type (nullable).

### Remarks
This property is part of the IProducerConfiguration interface in the KafkaFlow.Configuration namespace.
```

--------------------------------

### CreateIfNotExistsAsync(IEnumerable<TopicConfiguration>)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Clusters/IClusterManager/CreateIfNotExistsAsync/

Creates topics when building cluster configuration if the topic doesn't exist.

```APIDOC
## CreateIfNotExistsAsync(IEnumerable<TopicConfiguration>)

### Description
Creates topics when building cluster configuration if the topic doesn't exist.

### Method
CreateIfNotExistsAsync

### Signature
```csharp
public Task CreateIfNotExistsAsync(IEnumerable<TopicConfiguration> configuration)
```

### Parameters
- **configuration** (IEnumerable<TopicConfiguration>) - Topics to create

### Returns
Task
```

--------------------------------

### IConsumerConfiguration.PartitionsRevokedHandlers property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/IConsumerConfiguration/PartitionsRevokedHandlers/

Gets the handlers that will be called when the partitions are revoked.

```APIDOC
## IConsumerConfiguration.PartitionsRevokedHandlers property

### Description
Gets the handlers that will be called when the partitions are revoked.

### Property Signature
```csharp
public IReadOnlyList<Action<IDependencyResolver, List<TopicPartitionOffset>>> 
    PartitionsRevokedHandlers { get; }
```

### Property Type
- **IReadOnlyList<Action<IDependencyResolver, List<TopicPartitionOffset>>>** - A read-only list of actions that take an IDependencyResolver and a list of TopicPartitionOffset.

### See Also
- interface IConsumerConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### ClusterConfiguration.Consumers property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/ClusterConfiguration/Consumers/

Gets the list of consumers configured for the cluster.

```APIDOC
## ClusterConfiguration.Consumers property

### Description
Gets the list of consumers.

### Property Signature
```csharp
public IReadOnlyCollection<IConsumerConfiguration> Consumers { get; }
```

### Property Value
- **Type**: IReadOnlyCollection<IConsumerConfiguration>
- **Description**: The list of consumers.

### See Also
- interface IConsumerConfiguration
- class ClusterConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### EnableTelemetry(this IClusterConfigurationBuilder cluster, string topicName, string consumerGroup, TimeSpan? cleanRunInterval, TimeSpan? expiryTime, int topicPartition)

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow.Admin/KafkaFlow/ClusterConfigurationBuilderExtensions/EnableTelemetry/

Creates the telemetry producer and consumer to send and receive metric messages. This overload provides additional configuration for consumer group, cleanup interval, expiry time, and topic partition.

```APIDOC
## EnableTelemetry(this IClusterConfigurationBuilder cluster, string topicName, string consumerGroup, TimeSpan? cleanRunInterval, TimeSpan? expiryTime, int topicPartition)

### Description
Creates the telemetry producer and consumer to send and receive metric messages.

### Method
EnableTelemetry

### Parameters
#### Parameters
- **cluster** (IClusterConfigurationBuilder) - Required - The cluster configuration builder
- **topicName** (string) - Required - The topic to be used by the metric commands
- **consumerGroup** (string) - Required - The consumer group prefix
- **cleanRunInterval** (TimeSpan?) - Optional - How often run storage cleanup. Every 10 minutes by default
- **expiryTime** (TimeSpan?) - Optional - Cleanup will remove metrics older than specified interval. 6 hours by default
- **topicPartition** (int) - Optional - The partition used to produce and consumer telemetry data

### Return Value
IClusterConfigurationBuilder

### See Also
- class ClusterConfigurationBuilderExtensions
- namespace KafkaFlow
```

--------------------------------

### ResumeConsumerByName()

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow.Admin/KafkaFlow.Admin.Messages/ResumeConsumerByName/

The default constructor for the ResumeConsumerByName class. It initializes a new instance of the class without any parameters.

```APIDOC
## ResumeConsumerByName()

### Description
Initializes a new instance of the ResumeConsumerByName class using the default constructor.

### Method
Constructor

### Signature
```csharp
public ResumeConsumerByName()
```

### Parameters
None.

### Returns
A new instance of the ResumeConsumerByName class.
```

--------------------------------

### ClusterConfiguration.GetSecurityInformation()

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow/KafkaFlow.Configuration/ClusterConfiguration/GetSecurityInformation/

Gets the Kafka security information for the cluster configuration.

```APIDOC
## GetSecurityInformation()

### Description
Gets the kafka security information.

### Method
GetSecurityInformation

### Signature
```csharp
public SecurityInformation GetSecurityInformation()
```

### Returns
- **SecurityInformation** - The kafka security information.

### See Also
- class ClusterConfiguration
- namespace KafkaFlow.Configuration
```

--------------------------------

### Configure KafkaFlow with Schema Registry and Avro Serializer Middlewares

Source: https://farfetch.github.io/kafkaflow/docs/guides/middlewares/serializer-middleware/

C# example showing how to configure KafkaFlow with a schema registry broker and add Avro serializer/deserializer middlewares to producers and consumers. Requires the KafkaFlow.SchemaRegistry and KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro packages. Note that ConfluentJson type resolver can only resolve a single type per topic, and for multiple types per topic, use SubjectNameStrategy.Record or TopicRecord.

```csharp
public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddKafka(
            kafka => kafka
                .AddCluster(
                    cluster => cluster
                        .WithBrokers(new[] { "localhost:9092" })
                        .WithSchemaRegistry(config => config.Url = "localhost:8081")
                        .AddProducer(
                            ...
                            .AddMiddlewares(middlewares => 
                                    middlewares.AddSchemaRegistryAvroSerializer(new AvroSerializerConfig{ SubjectNameStrategy = SubjectNameStrategy.TopicRecord })
                        )
                       .AddConsumer(
                            ...
                            .AddMiddlewares(middlewares => middlewares.AddSchemaRegistryAvroDeserializer()
                        )
                    )
            );
    }
}
```

--------------------------------

### StopConsumerByName.ConsumerName property

Source: https://farfetch.github.io/kafkaflow/docs/reference/KafkaFlow.Admin/KafkaFlow.Admin.Messages/StopConsumerByName/ConsumerName/

Gets or sets the consumer name that will be stopped.

```APIDOC
## StopConsumerByName.ConsumerName property

### Description
Gets or sets the consumer name that will be stopped.

### Property Signature
```csharp
public string ConsumerName { get; set; }
```

### Property Value
- **Type**: string
- **Access**: Read/Write

### Remarks
This property is part of the `StopConsumerByName` class in the `KafkaFlow.Admin.Messages` namespace.

### See Also
- class `StopConsumerByName`
- namespace `KafkaFlow.Admin.Messages`
```
Tokens:
Raw