Remove cancellation message handler

paket.dependencies

@@ -90,10 +90,10 @@ nuget Be.Vlaanderen.Basisregisters.GrAr.Provenance.AcmIdm 21.14.1
 nuget Be.Vlaanderen.Basisregisters.GrAr.Extracts 21.14.1
 nuget Be.Vlaanderen.Basisregisters.GrAr.Oslo 21.14.1
 
-nuget Be.Vlaanderen.Basisregisters.MessageHandling.AwsSqs.Simple 5.0.1
-nuget Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Simple 5.0.1
-nuget Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Producer 5.0.1
-nuget Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer.SqlServer 5.0.1
+nuget Be.Vlaanderen.Basisregisters.MessageHandling.AwsSqs.Simple 5.2.0
+nuget Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Simple 5.2.0
+nuget Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Producer 5.2.0
+nuget Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer.SqlServer 5.2.0
 
 nuget Be.Vlaanderen.Basisregisters.Shaperon 10.0.2
 

paket.lock

@@ -305,19 +305,19 @@ NUGET
       Be.Vlaanderen.Basisregisters.GrAr.Common (21.14.1)
       Be.Vlaanderen.Basisregisters.GrAr.Provenance (21.14.1)
       Microsoft.CSharp (>= 4.7)
-    Be.Vlaanderen.Basisregisters.MessageHandling.AwsSqs.Simple (5.0.1)
+    Be.Vlaanderen.Basisregisters.MessageHandling.AwsSqs.Simple (5.2)
       AWSSDK.Core (>= 3.7.302.15)
       AWSSDK.SQS (>= 3.7.300.54)
       Microsoft.Extensions.Logging (>= 8.0)
-    Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer.SqlServer (5.0.1)
+    Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer.SqlServer (5.2)
       Confluent.Kafka (>= 2.3)
       Microsoft.EntityFrameworkCore.SqlServer (>= 8.0.2)
       Microsoft.Extensions.Logging (>= 8.0)
       Newtonsoft.Json (>= 13.0.3)
-    Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Producer (5.0.1)
+    Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Producer (5.2)
       Confluent.Kafka (>= 2.3)
       Newtonsoft.Json (>= 13.0.3)
-    Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Simple (5.0.1)
+    Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Simple (5.2)
       Confluent.Kafka (>= 2.3)
       Microsoft.Extensions.Logging (>= 8.0)
     Be.Vlaanderen.Basisregisters.Middleware.AddProblemJsonHeader (3.0)

src/ParcelRegistry.Consumer.Address.Console/Program.cs

@@ -13,6 +13,7 @@ namespace ParcelRegistry.Consumer.Address.Console
     using Be.Vlaanderen.Basisregisters.EventHandling;
     using Be.Vlaanderen.Basisregisters.MessageHandling.Kafka;
     using Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer;
+    using Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer.Extensions;
     using Destructurama;
     using Microsoft.EntityFrameworkCore;
     using Microsoft.Extensions.Configuration;
@@ -84,7 +85,7 @@ public static async Task Main(string[] args)
                             ));
 
                     services
-                        .AddDbContextFactory<ConsumerAddressContext>((provider, options) => options
+                        .AddDbContextFactory<ConsumerAddressContext>((_, options) => options
                             .UseLoggerFactory(loggerFactory)
                             .UseSqlServer(hostContext.Configuration.GetConnectionString("ConsumerAddress"), sqlServerOptions =>
                             {
@@ -118,25 +119,8 @@ public static async Task Main(string[] args)
                             hostContext.Configuration["Kafka:SaslUserName"],
                             hostContext.Configuration["Kafka:SaslPassword"]));
 
-                        var offset = hostContext.Configuration["AddressTopicOffset"];
-
-                        if (!string.IsNullOrWhiteSpace(offset) && long.TryParse(offset, out var result))
-                        {
-                            var ignoreDataCheck = hostContext.Configuration.GetValue<bool>("IgnoreAddressTopicOffsetDataCheck", false);
-
-                            if (!ignoreDataCheck)
-                            {
-                                using var ctx = c.Resolve<ConsumerAddressContext>();
-
-                                if (ctx.AddressConsumerItems.Any())
-                                {
-                                    throw new InvalidOperationException(
-                                        $"Cannot set Kafka offset to {offset} because {nameof(ctx.AddressConsumerItems)} has data.");
-                                }
-                            }
-
-                            consumerOptions.ConfigureOffset(new Offset(result));
-                        }
+                        using var ctx = c.Resolve<ConsumerAddressContext>();
+                        ctx.OverrideConfigureOffset(consumerOptions);
 
                         return consumerOptions;
                     });

src/ParcelRegistry.Consumer.Address.Console/appsettings.json

@@ -17,8 +17,6 @@
 
   "GroupSuffix": "",
   "AddressTopic": "dev.address",
-  "AddressTopicOffset": "",
-  "IgnoreAddressTopicOffsetDataCheck": false,
 
   "BaseUrl": "https://api.staging-basisregisters.vlaanderen/",
 

src/ParcelRegistry.Consumer.Address/BackOfficeConsumer.cs

@@ -57,14 +57,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             {
                 await _kafkaIdemIdompotencyConsumer.ConsumeContinuously(async (message, context) =>
                 {
-                    _logger.LogInformation("Handling next message");
-
-                    await commandHandlingProjector.ProjectAsync(commandHandler, message, stoppingToken).ConfigureAwait(false);
-                    await backOfficeProjector.ProjectAsync(context, message, stoppingToken).ConfigureAwait(false);
-
-                    //CancellationToken.None to prevent halfway consumption
-                    await context.SaveChangesAsync(CancellationToken.None);
-
+                    await HandleMessage(commandHandlingProjector, commandHandler, message, backOfficeProjector, context);
                 }, stoppingToken);
             }
             catch (Exception)
@@ -73,5 +66,17 @@ await _kafkaIdemIdompotencyConsumer.ConsumeContinuously(async (message, context)
                 throw;
             }
         }
+
+        private async Task HandleMessage(ConnectedProjector<CommandHandler> commandHandlingProjector, CommandHandler commandHandler, object message,
+            ConnectedProjector<ConsumerAddressContext> backOfficeProjector, ConsumerAddressContext context)
+        {
+            _logger.LogInformation("Handling next message");
+
+            await commandHandlingProjector.ProjectAsync(commandHandler, message, CancellationToken.None).ConfigureAwait(false);
+            await backOfficeProjector.ProjectAsync(context, message, CancellationToken.None).ConfigureAwait(false);
+
+            //CancellationToken.None to prevent halfway consumption
+            await context.SaveChangesAsync(CancellationToken.None);
+        }
     }
 }

src/ParcelRegistry.Consumer.Address/ConsumerAddressContext.cs

@@ -5,6 +5,7 @@ namespace ParcelRegistry.Consumer.Address
     using System.IO;
     using System.Linq;
     using System.Reflection;
+    using Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer;
     using Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer.SqlServer;
     using Be.Vlaanderen.Basisregisters.ProjectionHandling.Runner.SqlServer.MigrationExtensions;
     using Microsoft.EntityFrameworkCore;
@@ -15,11 +16,12 @@ namespace ParcelRegistry.Consumer.Address
     using Parcel.DataStructures;
     using ParcelRegistry.Infrastructure;
 
-    public class ConsumerAddressContext : SqlServerConsumerDbContext<ConsumerAddressContext>, IAddresses
+    public class ConsumerAddressContext : SqlServerConsumerDbContext<ConsumerAddressContext>, IAddresses, IOffsetOverrideDbSet
     {
         public override string ProcessedMessagesSchema => Schema.ConsumerAddress;
 
-        public DbSet<AddressConsumerItem> AddressConsumerItems { get; set; }
+        public DbSet<AddressConsumerItem> AddressConsumerItems => Set<AddressConsumerItem>();
+        public DbSet<OffsetOverride> OffsetOverrides => Set<OffsetOverride>();
 
         // This needs to be here to please EF
         public ConsumerAddressContext()
@@ -90,6 +92,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
 
             modelBuilder
                 .ApplyConfigurationsFromAssembly(typeof(ConsumerAddressContext).GetTypeInfo().Assembly);
+            modelBuilder.ApplyConfiguration(new OffsetOverrideConfiguration(ProcessedMessagesSchema));
         }
     }
 

src/ParcelRegistry.Consumer.Address/Migrations/20241205082405_AddOffsetOverride.cs

@@ -0,0 +1,47 @@
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace ParcelRegistry.Consumer.Address.Migrations
+{
+    /// <inheritdoc />
+    public partial class AddOffsetOverride : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.CreateTable(
+                name: "OffsetOverrides",
+                schema: "ParcelRegistryConsumerAddress",
+                columns: table => new
+                {
+                    ConsumerGroupId = table.Column<string>(type: "nvarchar(450)", nullable: false),
+                    Offset = table.Column<long>(type: "bigint", nullable: false),
+                    Configured = table.Column<bool>(type: "bit", nullable: false)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_OffsetOverrides", x => x.ConsumerGroupId);
+                });
+
+            migrationBuilder.CreateIndex(
+                name: "IX_Addresses_Position",
+                schema: "ParcelRegistryConsumerAddress",
+                table: "Addresses",
+                column: "Position");
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropTable(
+                name: "OffsetOverrides",
+                schema: "ParcelRegistryConsumerAddress");
+
+            migrationBuilder.DropIndex(
+                name: "IX_Addresses_Position",
+                schema: "ParcelRegistryConsumerAddress",
+                table: "Addresses");
+        }
+    }
+}

src/ParcelRegistry.Consumer.Address/Migrations/ConsumerAddressContextModelSnapshot.cs

@@ -18,10 +18,26 @@ protected override void BuildModel(ModelBuilder modelBuilder)
         {
 #pragma warning disable 612, 618
             modelBuilder
-                .HasAnnotation("ProductVersion", "6.0.3")
+                .HasAnnotation("ProductVersion", "8.0.3")
                 .HasAnnotation("Relational:MaxIdentifierLength", 128);
 
-            SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder, 1L, 1);
+            SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder);
+
+            modelBuilder.Entity("Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer.OffsetOverride", b =>
+                {
+                    b.Property<string>("ConsumerGroupId")
+                        .HasColumnType("nvarchar(450)");
+
+                    b.Property<bool>("Configured")
+                        .HasColumnType("bit");
+
+                    b.Property<long>("Offset")
+                        .HasColumnType("bigint");
+
+                    b.HasKey("ConsumerGroupId");
+
+                    b.ToTable("OffsetOverrides", "ParcelRegistryConsumerAddress");
+                });
 
             modelBuilder.Entity("Be.Vlaanderen.Basisregisters.MessageHandling.Kafka.Consumer.ProcessedMessage", b =>
                 {
@@ -76,6 +92,8 @@ protected override void BuildModel(ModelBuilder modelBuilder)
 
                     b.HasIndex("IsRemoved");
 
+                    b.HasIndex("Position");
+
                     b.ToTable("Addresses", "ParcelRegistryConsumerAddress");
                 });
 #pragma warning restore 612, 618