KafkaFlow Retry Extensions
KafkaFlow Retry is a .NET framework to implement easy resilience on consumers.
KafkaFlow Retry is an extension of Kafka Flow.
Resilience policies
| Policy | Description | Aka | Required Packages |
|---|---|---|---|
| Simple Retry (policy family) (quickstart ; deep) |
Many faults are transient and may self-correct after a short delay. | "Maybe it's just a blip" | KafkaFlow.Retry |
| Forever Retry (policy family) (quickstart ; deep) |
Many faults are semi-transient and may self-correct after multiple retries. | "Never give up" | KafkaFlow.Retry |
| Durable Retry (quickstart ; deep) |
Beyond a certain amount of retries and wait, you want to keep processing next-in-line messages but you can't loss the current offset message. As persistance databases, MongoDb or SqlServer are available. And you can manage in-retry messages through HTTP API. | "I can't stop processing messages but I can't loss messages" | KafkaFlow.Retry KafkaFlow.Retry.API KafkaFlow.Retry.SqlServer or KafkaFlow.Retry.MongoDb |
Installing via NuGet
Install packages related to your context. The Core package is required for all other packages.
Requirements
.NET Core 2.1 and later using Hosted Service
Packages
| Name | nuget.org |
|---|---|
| KafkaFlow.Retry |   |
| KafkaFlow.Retry.API |   |
| KafkaFlow.Retry.MongoDb |   |
| KafkaFlow.Retry.SqlServer |   |
Core package
Install-Package KafkaFlow.Retry
HTTP API package
Install-Package KafkaFlow.Retry.API
MongoDb package
Install-Package KafkaFlow.Retry.MongoDb
SqlServer package
Install-Package KafkaFlow.Retry.SqlServer
Usage – Simple and Forever retries policies
Simple
.AddMiddlewares(
middlewares => middlewares // KafkaFlow middlewares
.RetrySimple(
(config) => config
.Handle<ExternalGatewayException>() // Exceptions to be handled
.TryTimes(3)
.WithTimeBetweenTriesPlan((retryCount) =>
TimeSpan.FromMilliseconds(Math.Pow(2, retryCount)*1000) // exponential backoff
)
)Forever
.AddMiddlewares(
middlewares => middlewares // KafkaFlow middlewares
.RetryForever(
(config) => config
.Handle<DatabaseTimeoutException>() // Exceptions to be handled
.WithTimeBetweenTriesPlan(
TimeSpan.FromMilliseconds(500),
TimeSpan.FromMilliseconds(1000)
)
)
Usage – Durable retry policy
Durable
.AddMiddlewares(
middlewares => middlewares // KafkaFlow middlewares
.RetryDurable(
config => config
.Handle<NonBlockingException>() // 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<Handler>()
).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)
)
)
)See the setup page and samples for more details
Documentation
Contributing
Read the Contributing guidelines
Maintainers
- Bruno Gomes
- Carlos Miranda
- Fernando Marins
- Leandro Magalhães
- Luís Garcês
- Martinho Novais
- Rodrigo Belo
- Sérgio Ribeiro