### 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() .AddTypedHandlers(handlers => handlers .AddHandler()) ) ) .AddProducer("producer-name", producer => producer .DefaultTopic("sample-topic") .AddMiddlewares(middlewares => middlewares .AddSerializer() ) ) ) ); } 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(); 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 delegate. ```APIDOC ## ClusterConfiguration.OnStartedHandler property ### Description Gets the handler to be executed when the cluster started. ### Property Signature ```csharp public Action OnStartedHandler { get; } ``` ### Property Value - **Type**: `Action` - **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 Topics { get; } ``` ### Returns - **IReadOnlyList** - 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 All { get; } ``` ### Property Value - **Type**: IEnumerable - **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 RunningPartitions { get; } ``` ### Return Value - **IEnumerable** - 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>> PartitionsAssignedHandlers { get; } ``` ### Property Value - **Type**: IReadOnlyList>> - **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 Topics { get; set; } ``` ### Property Value - **Type**: IList - **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 MiddlewaresConfigurations { get; } ``` ### Property Value - **Type**: IReadOnlyList - **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() ) ) ) ); var serviceProvider = services.BuildServiceProvider(); var producer = serviceProvider .GetRequiredService() .GetProducer(producerName); await producer.ProduceAsync( topicName, Guid.NewGuid().ToString(), new HelloMessage { Text = "Hello!" }); Console.WriteLine("Message sent!"); ``` -------------------------------- ### PartitionKeyDistributionStrategy.Initialize(IReadOnlyList) 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) ### Description Initializes the distribution strategy with the provided list of workers. ### Method Initialize ### Signature public void Initialize(IReadOnlyList workers) ### Parameters - **workers** (IReadOnlyList) - 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() 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() .AddTypedHandlers(h => h.AddHandler()) ) ) ) ); 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> GetEnumerator() ### Return Type IEnumerator> ### 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 Topics { get; } ``` ### Property Value - **Type**: IReadOnlyList - **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 { 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 Brokers { get; } ``` ### Property Value - **Type**: IReadOnlyCollection - **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 Clusters { get; } ``` ### Property Value - **Type**: IReadOnlyCollection - **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(producer => producer ... .AddMiddlewares(middlewares => middlewares ... .AddCompressor() // 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() ) ) ``` -------------------------------- ### 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 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 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 partitions) ``` ### Parameters - **name** (string) - Required - The name of the topic. - **partitions** (IReadOnlyCollection) - Required - A collection of partition metadata for the topic. ### See Also - class TopicPartitionMetadata - record TopicMetadata - namespace KafkaFlow ``` -------------------------------- ### GetProducer() 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() ### Description Gets a producer by its type. ### Method GetProducer ### 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() .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() .AddSerializer() ) ) ``` -------------------------------- ### 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 Configs { get; } ``` ### Property Value - **Type**: `Dictionary` - **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 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) - Required - Additional topic creation configuration values. ### Request Example ```csharp public TopicConfiguration(string name, int partitions, short replicas, Dictionary 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 PausedPartitions { get; } ``` ### Property Type - **IReadOnlyList** - 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(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() .AddDeserializer() ) ) ``` -------------------------------- ### 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) 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) ### Description Creates topics when building cluster configuration if the topic doesn't exist. ### Method CreateIfNotExistsAsync ### Signature ```csharp public Task CreateIfNotExistsAsync(IEnumerable configuration) ``` ### Parameters - **configuration** (IEnumerable) - 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>> PartitionsRevokedHandlers { get; } ``` ### Property Type - **IReadOnlyList>>** - 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 Consumers { get; } ``` ### Property Value - **Type**: IReadOnlyCollection - **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` ```