### Start Local Development Server Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/README.md Starts a local development server for live preview. Changes are reflected without restarting. ```bash yarn start ``` -------------------------------- ### Install KafkaFlow.Retry Package Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/simple-retries.md Install the KafkaFlow.Retry NuGet package using the .NET CLI. ```bash dotnet add package KafkaFlow.Retry ``` -------------------------------- ### Install Dependencies Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/README.md Installs project dependencies using Yarn. ```bash yarn ``` -------------------------------- ### Start Kafka Cluster with Docker Compose Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Starts a local Apache Kafka cluster using Docker. Ensure Docker Desktop is running. ```bash docker-compose up -d ``` -------------------------------- ### Install KafkaFlow Packages for Producer Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Installs necessary KafkaFlow and related packages for the producer application. ```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 Retry SQL Server Provider Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Install the KafkaFlow.Retry.SqlServer NuGet package to use SQL Server as the durable retry data provider. ```bash dotnet add package KafkaFlow.Retry.SqlServer ``` -------------------------------- ### Install KafkaFlow Retry MongoDB Provider Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Install the KafkaFlow.Retry.MongoDb NuGet package to use MongoDB as the durable retry data provider. ```bash dotnet add package KafkaFlow.Retry.MongoDb ``` -------------------------------- ### Add KafkaFlow.Retry.Postgres Package Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Install the KafkaFlow.Retry.Postgres package using the .NET CLI to enable durable retries with a PostgreSQL data provider. ```bash dotnet add package KafkaFlow.Retry.Postgres ``` -------------------------------- ### Add KafkaFlow.Retry.API Package Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Install the KafkaFlow.Retry.API package to enable an HTTP API for managing the retry data provider. ```bash dotnet add package KafkaFlow.Retry.API ``` -------------------------------- ### Configure Simple Retries Middleware Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/simple-retries.md Add the RetrySimple middleware to your KafkaFlow consumer configuration. This example demonstrates handling a specific exception type and configuring exponential backoff for retries. ```csharp .AddMiddlewares( middlewares => middlewares // KafkaFlow middlewares .RetrySimple( (config) => config .Handle() // Exceptions to be handled .TryTimes(3) .WithTimeBetweenTriesPlan((retryCount) => TimeSpan.FromMilliseconds(Math.Pow(2, retryCount) * 1000) // exponential backoff ) ) ... ) ``` -------------------------------- ### Configure Simple Retry Middleware Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/installation.md Register the simple retry middleware in your KafkaFlow configuration. This example handles `ExternalGatewayException`, retries 3 times, and uses an exponential backoff strategy. ```csharp .AddMiddlewares( middlewares => middlewares // KafkaFlow middlewares .RetrySimple( (config) => config .Handle() // Exception to be handled .TryTimes(3) .WithTimeBetweenTriesPlan((retryCount) => TimeSpan.FromMilliseconds(Math.Pow(2, retryCount)*1000) // exponential backoff ) ) ``` -------------------------------- ### Configure Durable Retries Middleware Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Configure the RetryDurable middleware within KafkaFlow's configuration. This example shows how to set up retry policies, message types, embedded retry clusters, queue polling jobs, and data providers. ```csharp .AddMiddlewares( middlewares => middlewares // KafkaFlow middlewares .RetryDurable( (config) => config .Handle() .WithMessageType(typeof(OrderMessage)) // Message type to be consumed .WithEmbeddedRetryCluster( // Retry consumer config cluster, config => config .WithRetryTopicName("order-topic-retry") .WithRetryConsumerBufferSize(4) .WithRetryConsumerWorkersCount(2) .WithRetryConsumerStrategy(RetryConsumerStrategy.GuaranteeOrderedConsumption) .WithRetryTypedHandlers( handlers => handlers .WithHandlerLifetime(InstanceLifetime.Transient) .AddHandler() ).Enabled(true) ) .WithQueuePollingJobConfiguration( // Polling configuration config => config .WithId("custom_search_key") .WithCronExpression("0 0/1 * 1/1 * ? *") .WithExpirationIntervalFactor(1) .WithFetchSize(10) .Enabled(true) ) .WithMongoDbDataProvider(...) // OR .WithPostgresDataProvider(...) // OR .WithSqlServerDataProvider(...) .WithRetryPlanBeforeRetryDurable( // Chained simple retry before triggering durable config => config .TryTimes(3) .WithTimeBetweenTriesPlan( TimeSpan.FromMilliseconds(250), TimeSpan.FromMilliseconds(500), TimeSpan.FromMilliseconds(1000)) .ShouldPauseConsumer(false) ) ) ... ) ``` -------------------------------- ### Run the Consumer Application Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Command to execute the consumer project from the project directory. ```bash dotnet run --project Consumer/Consumer.csproj ``` -------------------------------- ### Create Consumer Project with .NET CLI Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Creates a new .NET console application for the Kafka consumer. ```bash dotnet new console --name Consumer ``` -------------------------------- ### Build Static Website Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/README.md Generates static content for deployment to any hosting service. ```bash yarn build ``` -------------------------------- ### Run the Producer Application Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Command to execute the producer project from the project directory. ```bash dotnet run --project Producer/Producer.csproj ``` -------------------------------- ### Create Producer Project with .NET CLI Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Creates a new .NET console application for the Kafka producer. ```bash dotnet new console --name Producer ``` -------------------------------- ### Deploy Website (SSH) Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/README.md Builds and deploys the website using SSH, typically for GitHub Pages. ```bash USE_SSH=true yarn deploy ``` -------------------------------- ### Configure and Send Message with KafkaFlow Producer Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Sets up KafkaFlow, configures a producer with JSON serialization, and sends a message to a Kafka topic. ```csharp using Microsoft.Extensions.DependencyInjection; using KafkaFlow.Producers; using KafkaFlow; using KafkaFlow.Serializer; 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!"); ``` -------------------------------- ### Deploy Website (No SSH) Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/README.md Builds and deploys the website without SSH, requiring your GitHub username. ```bash GIT_USER= yarn deploy ``` -------------------------------- ### Git Branching and Committing Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/CONTRIBUTING.md Follow these commands to create a new feature branch, commit your changes, and push them to the repository. ```bash git checkout -b my-new-feature ``` ```bash git commit -am 'feat: Add some feature' ``` ```bash git push origin my-new-feature ``` -------------------------------- ### Configure Avro Producer and Consumer with Retry Source: https://github.com/farfetch/kafkaflow-retry-extensions/wiki/Serializer-Middleware Sets up a Kafka producer with Avro serialization and a consumer with Avro serialization and durable retry handling. Ensure AvroSerializerConfig is correctly configured for subject naming strategy. ```csharp internal static class KafkaClusterConfigurationBuilder { internal static IClusterConfigurationBuilder SetupRetryDurableMongoAvroDb( this IClusterConfigurationBuilder cluster, string mongoDbConnectionString, string mongoDbDatabaseName, string mongoDbRetryQueueCollectionName, string mongoDbRetryQueueItemCollectionName) { cluster.AddProducer( "kafka-flow-retry-durable-mongodb-avro-producer", producer => producer .DefaultTopic("sample-kafka-flow-retry-durable-mongodb-avro-topic") .WithCompression(Confluent.Kafka.CompressionType.Gzip) .AddMiddlewares( middlewares => middlewares .AddSchemaRegistryAvroSerializer( new AvroSerializerConfig { SubjectNameStrategy = SubjectName.TopicRecord }) ) .WithAcks(Acks.All) ) .AddConsumer( consumer => consumer .Topic("sample-kafka-flow-retry-durable-mongodb-avro-topic") .WithGroupId("sample-consumer-kafka-flow-retry-durable-mongodb-avro") .WithName("kafka-flow-retry-durable-mongodb-avro-consumer") .WithBufferSize(10) .WithWorkersCount(20) .WithAutoOffsetReset(AutoOffsetReset.Latest) .AddMiddlewares( middlewares => middlewares .AddSchemaRegistryAvroSerializer() .RetryDurable( configure => configure .Handle() .WithMessageType(typeof(AvroLogMessage)) .WithMessageSerializeSettings(new JsonSerializerSettings { ContractResolver = new WritablePropertiesOnlyResolver() }) ... ); ``` -------------------------------- ### Import KafkaFlow.Retry Namespaces Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/installation.md Include these namespaces to access KafkaFlow and its retry functionalities in your C# code. ```csharp using KafkaFlow; using KafkaFlow.Retry; ``` -------------------------------- ### Configure Kafka Consumer with Retry Middleware Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Set up the KafkaFlow bus with a consumer that includes the RetrySimple middleware. This configuration specifies which exceptions to retry, the number of attempts, and the delay strategy between retries. ```csharp using KafkaFlow; using Microsoft.Extensions.DependencyInjection; using KafkaFlow.Retry; using KafkaFlow.Serializer; 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) .WithAutoOffsetReset(AutoOffsetReset.Earliest) .WithGroupId("sample-group") .WithBufferSize(100) .WithWorkersCount(10) .AddMiddlewares( middlewares => middlewares .RetrySimple( (config) => config .Handle() .TryTimes(3) .WithTimeBetweenTriesPlan((retryCount) => TimeSpan.FromMilliseconds(Math.Pow(2, retryCount) * 1000) ) ) .AddDeserializer() .AddTypedHandlers(h => h.AddHandler()) ) ) ) ); var serviceProvider = services.BuildServiceProvider(); var bus = serviceProvider.CreateKafkaBus(); await bus.StartAsync(); Console.ReadKey(); await bus.StopAsync(); ``` -------------------------------- ### Handle Any Exception Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/exception-handling.md Configure the retry policy to handle any exception regardless of its type by using `HandleAnyException()`. This is a broad approach suitable when any failure should be retried. ```csharp .AddMiddlewares( middlewares => middlewares .RetrySimple( (config) => config .HandleAnyException() ... ) ... ) ``` -------------------------------- ### Configure Simple Retry Policy Source: https://github.com/farfetch/kafkaflow-retry-extensions/wiki/Setup Use this to configure a simple retry policy that attempts to re-process a message a fixed number of times with an exponential backoff strategy. ```csharp .AddMiddlewares( middlewares => middlewares // KafkaFlow middlewares .RetrySimple( (config) => config .Handle() // Exceptions to be handled .TryTimes(3) .WithTimeBetweenTriesPlan((retryCount) => TimeSpan.FromMilliseconds(Math.Pow(2, retryCount)*1000) // exponential backoff ) ) ``` -------------------------------- ### Configure Forever Retry Policy Source: https://github.com/farfetch/kafkaflow-retry-extensions/wiki/Setup Configure a retry policy that will continue to retry indefinitely until successful, using a predefined plan for time between retries. ```csharp .AddMiddlewares( middlewares => middlewares // KafkaFlow middlewares .RetryForever( (config) => config .Handle() // Exceptions to be handled .WithTimeBetweenTriesPlan( TimeSpan.FromMilliseconds(500), TimeSpan.FromMilliseconds(1000) ) ) ``` -------------------------------- ### Configure Postgres Data Provider for Retries Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Configure the retry middleware to use PostgreSQL as the data provider by specifying the connection string and database name. ```csharp .AddMiddlewares( middlewares => middlewares .RetryDurable( (config) => config ... .WithPostgresDataProvider( connectionString, databaseName) ... ) ... ) ``` -------------------------------- ### Configure Durable Retry Policy Source: https://github.com/farfetch/kafkaflow-retry-extensions/wiki/Setup Set up a durable retry policy that persists retry information, allowing for recovery after application restarts. This includes configuring the retry consumer, polling job, persistence provider, and an optional chained simple retry. ```csharp .AddMiddlewares( middlewares => middlewares // KafkaFlow middlewares .RetryDurable( config => config .Handle() // Exceptions to be handled .WithMessageType(typeof(TestMessage)) // Message type to be consumed .WithEmbeddedRetryCluster( // Retry consumer config cluster, config => config .WithRetryTopicName("test-topic-retry") .WithRetryConsumerBufferSize(4) .WithRetryConsumerWorkersCount(2) .WithRetryConusmerStrategy(RetryConsumerStrategy.GuaranteeOrderedConsumption) .WithRetryTypedHandlers( handlers => handlers .WithHandlerLifetime(InstanceLifetime.Transient) .AddHandler() ).Enabled(true) ) .WithQueuePollingJobConfiguration( // Polling configuration config => config .WithId("custom_search_key") .WithCronExpression("0 0/1 * 1/1 * ? *") .WithExpirationIntervalFactor(1) .WithFetchSize(10) .Enabled(true) ) .WithMongoDbDataProvider( // Persistence configuration mongoDbconnectionString, mongoDbdatabaseName, mongoDbretryQueueCollectionName, mongoDbretryQueueItemCollectionName ) .WithRetryPlanBeforeRetryDurable( // Chained simple retry before triggering durable config => config .TryTimes(3) .WithTimeBetweenTriesPlan( TimeSpan.FromMilliseconds(250), TimeSpan.FromMilliseconds(500), TimeSpan.FromMilliseconds(1000)) .ShouldPauseConsumer(false) ) ) ) ``` -------------------------------- ### Configure RetryForever Middleware Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/forever-retries.md Add the RetryForever middleware to your KafkaFlow consumer configuration. Specify exceptions to handle and a plan for time between retries. ```csharp .AddMiddlewares( middlewares => middlewares // KafkaFlow middlewares .RetryForever( (config) => config .Handle() // Exceptions to be handled .WithTimeBetweenTriesPlan( TimeSpan.FromMilliseconds(500), TimeSpan.FromMilliseconds(1000) ) ) ... ) ``` -------------------------------- ### Add Project Reference to Producer Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Adds a project reference from the Consumer project to the Producer project to share message contracts. ```bash dotnet add reference ../Producer ``` -------------------------------- ### Pause Consumer on Exception Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/simple-retries.md Configure the simple retry policy to pause the consumer when a specified exception occurs by using the ShouldPauseConsumer method. ```csharp .AddMiddlewares( middlewares => middlewares .RetrySimple( (config) => config ... .ShouldPauseConsumer(true) ... ) ... ) ``` -------------------------------- ### Configure Durable Retries with SQL Server Provider Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Configure durable retries to use SQL Server as the data provider by specifying the connection string and database name. ```csharp .AddMiddlewares( middlewares => middlewares .RetryDurable( (config) => config ... .WithSqlServerDataProvider( connectionString, databaseName) ... ) ... ) ``` -------------------------------- ### Configure Durable Retries with MongoDB Provider Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Configure durable retries to use MongoDB as the data provider by specifying connection details and collection names. ```csharp .AddMiddlewares( middlewares => middlewares .RetryDurable( (config) => config ... .WithMongoDbDataProvider( connectionString, database, retryQueueCollectionName, retryQueueItemCollectionName) ... ) ... ) ``` -------------------------------- ### Configure Durable Retries with Avro Serialization Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/durable-retries.md Configure durable retries with Avro serialization, specifying the message type and serializer settings. Ensure the Avro serializer middleware is added. ```csharp .AddMiddlewares( middlewares => middlewares .AddSchemaRegistryAvroSerializer() .RetryDurable( (config) => config .Handle() .WithMessageType(typeof(AvroLogMessage)) .WithMessageSerializeSettings(new JsonSerializerSettings { ContractResolver = new WritablePropertiesOnlyResolver() }) ... ) ... ) ``` -------------------------------- ### Handle Specific Exception Type Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/exception-handling.md Configure the retry policy to handle a specific exception type by using `Handle()`. This is useful when only certain exceptions should trigger a retry. ```csharp .AddMiddlewares( middlewares => middlewares .RetrySimple( (config) => config .Handle() // Exceptions to be handled ... ) ... ) ``` -------------------------------- ### Handle Exceptions Based on Criteria Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/guides/exception-handling.md Handle multiple exception types or apply complex conditions using `Handle(Func)`. This allows for granular control over which exceptions trigger a retry based on custom logic. ```csharp .AddMiddlewares( middlewares => middlewares .RetrySimple( (config) => config .Handle(context => context.Exception is ArgumentException or ArgumentNullException) ... ) ... ) ``` -------------------------------- ### Implement a Message Handler with Retry Logic Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Define a message handler that can optionally throw an exception to simulate failure. This handler is designed to be used with KafkaFlow's retry mechanism. ```csharp using KafkaFlow; using Producer; namespace Consumer; public class HelloMessageHandler : IMessageHandler { private static readonly Random Random = new(Guid.NewGuid().GetHashCode()); private static bool ShouldFail() => Random.Next(2) == 1; public Task Handle(IMessageContext context, HelloMessage message) { if (ShouldFail()) { Console.WriteLine( "Let's fail: Partition: {0} | Offset: {1} | Message: {2}", context.ConsumerContext.Partition, context.ConsumerContext.Offset, message.Text); throw new IOException(); } Console.WriteLine( "Partition: {0} | Offset: {1} | Message: {2}", context.ConsumerContext.Partition, context.ConsumerContext.Offset, message.Text); return Task.CompletedTask; } } ``` -------------------------------- ### Define Message Contract Source: https://github.com/farfetch/kafkaflow-retry-extensions/blob/main/website/docs/getting-started/quickstart.md Defines a simple message contract for data serialization. ```csharp namespace Producer; public class HelloMessage { public string Text { get; set; } = default!; } ``` -------------------------------- ### Custom Contract Resolver for Serializer Source: https://github.com/farfetch/kafkaflow-retry-extensions/wiki/Serializer-Middleware Extends DefaultContractResolver to create a custom resolver that only includes writable properties. This is useful for controlling serialization behavior. ```csharp internal class WritablePropertiesOnlyResolver : DefaultContractResolver { protected override IList CreateProperties(Type type, MemberSerialization memberSerialization) { IList props = base.CreateProperties(type, memberSerialization); return props.Where(p => p.Writable).ToList(); } } ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.