Skip to content

Commit

Permalink
feat: delete queues cleanup (#100)
Browse files Browse the repository at this point in the history
  • Loading branch information
@fernando-a-marins
fernando-a-marins authored Mar 1, 2023
1 parent 1c406b3 commit 86daad2
Show file tree
Hide file tree
Showing 66 changed files with 1,808 additions and 726 deletions.
41 changes: 29 additions & 12 deletions samples/KafkaFlow.Retry.Sample/Helpers/KafkaClusterConfigurationBuilderHelper.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,13 +74,23 @@ internal static IClusterConfigurationBuilder SetupRetryDurableMongoDb(
)
.Enabled(true)
)
.WithQueuePollingJobConfiguration(
.WithPollingJobsConfiguration(
configure => configure
.WithId("retry-durable-mongodb-polling-id")
.WithCronExpression("0 0/1 * 1/1 * ? *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(10)
.Enabled(true)
.WithSchedulerId("retry-durable-mongodb-polling-id")
.WithRetryDurablePollingConfiguration(
configure => configure
.WithCronExpression("0 0/1 * 1/1 * ? *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(10)
.Enabled(true)
)
.WithCleanupPollingConfiguration(
configure => configure
.WithCronExpression("0 0 * 1/1 * ? *")
.WithRowsPerRequest(1048)
.WithTimeToLiveInDays(60)
.Enabled(true)
)
))
.AddTypedHandlers(
handlers => handlers
Expand Down Expand Up @@ -150,13 +160,20 @@ internal static IClusterConfigurationBuilder SetupRetryDurableSqlServer(
)
.Enabled(true)
)
.WithQueuePollingJobConfiguration(
.WithPollingJobsConfiguration(
configure => configure
.WithId("retry-durable-sqlserver-polling-id")
.WithCronExpression("0 0/1 * 1/1 * ? *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(10)
.Enabled(true)
.WithSchedulerId("retry-durable-sqlserver-polling-id")
.WithRetryDurablePollingConfiguration(
configure => configure
.WithCronExpression("0 0/1 * 1/1 * ? *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(10)
.Enabled(true)
)
.WithCleanupPollingConfiguration(
configure => configure
.Enabled(false)
)
))
.AddTypedHandlers(
handlers => handlers
Expand Down
15 changes: 9 additions & 6 deletions ...s/KafkaFlow.Retry.SchemaRegistry.Sample/Helpers/KafkaClusterConfigurationBuilderHelper.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,13 +84,16 @@ internal static IClusterConfigurationBuilder SetupRetryDurableMongoAvroDb(
)
.Enabled(true)
)
.WithQueuePollingJobConfiguration(
.WithPollingJobsConfiguration(
configure => configure
.WithId("retry-durable-mongodb-avro-polling-id")
.WithCronExpression("0 0/1 * 1/1 * ? *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(10)
.Enabled(true)
.WithSchedulerId("retry-durable-mongodb-avro-polling-id")
.WithRetryDurablePollingConfiguration(
configure => configure
.WithCronExpression("0 0/1 * 1/1 * ? *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(10)
.Enabled(true)
)
))
.AddTypedHandlers(
handlers => handlers
Expand Down
61 changes: 37 additions & 24 deletions src/KafkaFlow.Retry.IntegrationTests/Core/Bootstrappers/BootstrapperKafka.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -89,13 +89,16 @@ internal static IClusterConfigurationBuilder SetupRetryDurableGuaranteeOrderedCo
handlers => handlers
.WithHandlerLifetime(InstanceLifetime.Transient)
.AddHandler<RetryDurableTestMessageHandler>()))
.WithQueuePollingJobConfiguration(
.WithPollingJobsConfiguration(
configure => configure
.Enabled(true)
.WithId("custom_search_key_durable_guarantee_ordered_consumption_mongo_db")
.WithCronExpression("0/30 * * ? * * *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(256))
.WithSchedulerId("custom_search_key_durable_guarantee_ordered_consumption_mongo_db")
.WithRetryDurablePollingConfiguration(
configure => configure
.Enabled(true)
.WithCronExpression("0/30 * * ? * * *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(256))
)
.WithMongoDbDataProvider(
mongoDbConnectionString,
mongoDbDatabaseName,
Expand Down Expand Up @@ -162,13 +165,16 @@ internal static IClusterConfigurationBuilder SetupRetryDurableGuaranteeOrderedCo
handlers => handlers
.WithHandlerLifetime(InstanceLifetime.Transient)
.AddHandler<RetryDurableTestMessageHandler>()))
.WithQueuePollingJobConfiguration(
.WithPollingJobsConfiguration(
configure => configure
.Enabled(true)
.WithId("custom_search_key_durable_guarantee_ordered_consumption_sql_server")
.WithCronExpression("0/30 * * ? * * *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(256))
.WithSchedulerId("custom_search_key_durable_guarantee_ordered_consumption_sql_server")
.WithRetryDurablePollingConfiguration(
configure => configure
.Enabled(true)
.WithCronExpression("0/30 * * ? * * *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(256))
)
.WithSqlServerDataProvider(
sqlServerConnectionString,
sqlServerDatabaseName)
Expand Down Expand Up @@ -235,13 +241,16 @@ internal static IClusterConfigurationBuilder SetupRetryDurableLatestConsumptionM
handlers => handlers
.WithHandlerLifetime(InstanceLifetime.Transient)
.AddHandler<RetryDurableTestMessageHandler>()))
.WithQueuePollingJobConfiguration(
.WithPollingJobsConfiguration(
configure => configure
.Enabled(true)
.WithId("custom_search_key_durable_latest_consumption_mongo_db")
.WithCronExpression("0/30 * * ? * * *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(256))
.WithSchedulerId("custom_search_key_durable_latest_consumption_mongo_db")
.WithRetryDurablePollingConfiguration(
configure => configure
.Enabled(true)
.WithCronExpression("0/30 * * ? * * *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(256))
)
.WithMongoDbDataProvider(
mongoDbConnectionString,
mongoDbDatabaseName,
Expand Down Expand Up @@ -308,13 +317,17 @@ internal static IClusterConfigurationBuilder SetupRetryDurableLatestConsumptionS
handlers => handlers
.WithHandlerLifetime(InstanceLifetime.Transient)
.AddHandler<RetryDurableTestMessageHandler>()))
.WithQueuePollingJobConfiguration(
.WithPollingJobsConfiguration(
configure => configure
.Enabled(true)
.WithId("custom_search_key_durable_latest_consumption_sql_server")
.WithCronExpression("0/30 * * ? * * *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(256))
.WithSchedulerId("custom_search_key_durable_latest_consumption_sql_server")
.WithRetryDurablePollingConfiguration(
configure => configure
.Enabled(true)
.WithCronExpression("0/30 * * ? * * *")
.WithExpirationIntervalFactor(1)
.WithFetchSize(256))
)
.WithSqlServerDataProvider(
sqlServerConnectionString,
sqlServerDatabaseName)
Expand Down
3 changes: 1 addition & 2 deletions src/KafkaFlow.Retry.IntegrationTests/Core/Storages/Repositories/SqlServerRepository.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,8 +98,7 @@ public async Task CreateQueueAsync(RetryQueue queue)
DomainRetryQueueId = queue.Id,
Status = item.Status,
SeverityLevel = item.SeverityLevel,
Description = item.Description,
Sort = item.Sort // TODO: FIX-30: remove this after fix https://github.com/Farfetch/kafkaflow-retry-extensions/issues/30
Description = item.Description
};

var itemId = await this.retryQueueItemRepository.AddAsync(dbConnection, itemDbo);
Expand Down
Loading

0 comments on commit 86daad2

Please sign in to comment.