CxODEV-2125: keep the Kafka cancellation consumer running after a bad message (3.10.1 backport)

src/ConductorSharp.Client/ConductorSharp.Client.csproj

@@ -6,7 +6,7 @@
 		<Authors>Codaxy</Authors>
 		<Company>Codaxy</Company>
 		<PackageId>ConductorSharp.Client</PackageId>
-		<Version>3.10.0</Version>
+		<Version>3.10.1</Version>
 		<Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 		<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 		<PackageTags>netflix;conductor</PackageTags>

src/ConductorSharp.Engine/ConductorSharp.Engine.csproj

@@ -6,7 +6,7 @@
     <Authors>Codaxy</Authors>
 	<Company>Codaxy</Company>
 	<PackageId>ConductorSharp.Engine</PackageId>
-	<Version>3.10.0</Version>
+	<Version>3.10.1</Version>
     <Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 	<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 	<PackageTags>netflix;conductor</PackageTags>

src/ConductorSharp.KafkaCancellationNotifier/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("ConductorSharp.Engine.Tests")]

src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj

@@ -4,7 +4,7 @@
     <TargetFramework>net6.0</TargetFramework>
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
-    <Version>3.10.0</Version>
+    <Version>3.10.1</Version>
     <Authors>Codaxy</Authors>
     <Company>Codaxy</Company>
   </PropertyGroup>

src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaConsumerBackgroundService.cs

@@ -35,7 +35,6 @@ public TaskStatusModel Deserialize(ReadOnlySpan<byte> data, bool isNull, Seriali
         private readonly IOptions<KafkaOptions> _kafkaOptions;
         private readonly KafkaCancellationNotifier _notifier;
         private readonly ILogger<KafkaConsumerBackgroundService> _logger;
-        private const int KafkaRetryPeriodSeconds = 5;
 
         public KafkaConsumerBackgroundService(
             IOptions<KafkaOptions> kafkaOptions,
@@ -65,29 +64,28 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             consumer.Subscribe(_kafkaOptions.Value.TopicName);
 
             await Task.Run(
-                async () =>
+                () =>
                 {
                     try
                     {
                         while (true)
                         {
-                            var result = consumer.Consume(stoppingToken);
-                            _notifier.HandleKafkaEvent(result.Message.Value);
+                            // One bad message must not stop the consumer, so log it and move on to the next one.
+                            try
+                            {
+                                var result = consumer.Consume(stoppingToken);
+                                _notifier.HandleKafkaEvent(result.Message.Value);
+                            }
+                            catch (Exception e) when (e is not OperationCanceledException)
+                            {
+                                _logger.LogError(e, "Exception during message consumption from kafka, skipping the message");
+                            }
                         }
                     }
                     catch (OperationCanceledException)
                     {
                         _logger.LogInformation("Stopping KafkaCancellationNotifier background service");
                     }
-                    catch (Exception e)
-                    {
-                        _logger.LogError(
-                            e,
-                            "Exception during message consumption from kafka, will retry to consume after {Period} seconds",
-                            KafkaRetryPeriodSeconds
-                        );
-                        await Task.Delay(TimeSpan.FromSeconds(KafkaRetryPeriodSeconds), stoppingToken);
-                    }
                 },
                 stoppingToken
             );

src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj

@@ -7,7 +7,7 @@
     <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
     <Authors>Codaxy</Authors>
     <Company>Codaxy</Company>
-    <Version>3.10.0</Version>
+    <Version>3.10.1</Version>
   </PropertyGroup>
 
 	<ItemGroup>

test/ConductorSharp.Engine.Tests/ConductorSharp.Engine.Tests.csproj

@@ -94,6 +94,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\..\src\ConductorSharp.Engine\ConductorSharp.Engine.csproj" />
+    <ProjectReference Include="..\..\src\ConductorSharp.KafkaCancellationNotifier\ConductorSharp.KafkaCancellationNotifier.csproj" />
     <ProjectReference Include="..\..\src\ConductorSharp.Patterns\ConductorSharp.Patterns.csproj" />
   </ItemGroup>
 

test/ConductorSharp.Engine.Tests/Integration/KafkaCancellationNotifierTests.cs

@@ -0,0 +1,105 @@
+using ConductorSharp.KafkaCancellationNotifier;
+using ConductorSharp.KafkaCancellationNotifier.Service;
+using Confluent.Kafka;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using KafkaNotifier = ConductorSharp.KafkaCancellationNotifier.Service.KafkaCancellationNotifier;
+
+namespace ConductorSharp.Engine.Tests.Integration;
+
+public class KafkaCancellationNotifierTests : IAsyncLifetime
+{
+    private const string TaskName = "TEST_cancellable_task";
+    private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(20);
+
+    // librdkafka's in-process mock cluster speaks the real Kafka protocol, so no broker or Docker is needed.
+    private readonly IProducer<Null, string> _producer = new ProducerBuilder<Null, string>(
+        new Dictionary<string, string> { ["test.mock.num.brokers"] = "1" }
+    ).Build();
+    private readonly TopicPartition _partition = new("conductor.status.task", new Partition(0));
+    private readonly KafkaNotifier _notifier = new(new[] { new TaskToWorker { TaskName = TaskName } }, NullLogger<KafkaNotifier>.Instance);
+    private ICancellationNotifier.ICancellationTokenHolder _probe = null!;
+    private KafkaConsumerBackgroundService _service = null!;
+
+    public async Task InitializeAsync()
+    {
+        // Producing creates the topic, so the consumer subscribes to a topic that exists.
+        await _producer.ProduceAsync(_partition, StatusEvent("warm-up-task", "COMPLETED"));
+
+        using var admin = new DependentAdminClientBuilder(_producer.Handle).Build();
+        var bootstrapServers = string.Join(",", admin.GetMetadata(WaitTimeout).Brokers.Select(b => $"{b.Host}:{b.Port}"));
+
+        _service = new KafkaConsumerBackgroundService(
+            Options.Create(
+                new KafkaOptions
+                {
+                    BootstrapServers = bootstrapServers,
+                    TopicName = _partition.Topic,
+                    GroupId = "test"
+                }
+            ),
+            NullLogger<KafkaConsumerBackgroundService>.Instance,
+            _notifier
+        );
+        await _service.StartAsync(CancellationToken.None);
+
+        // The consumer starts at the latest offset, so repeat a probe event until one is handled.
+        _probe = _notifier.GetCancellationToken("probe-task", CancellationToken.None);
+        var deadline = DateTime.UtcNow + WaitTimeout;
+        do
+        {
+            Assert.True(DateTime.UtcNow < deadline, "The consumer did not start consuming");
+            await _producer.ProduceAsync(_partition, StatusEvent("probe-task", "CANCELED"));
+        } while (!_probe.CancellationToken.WaitHandle.WaitOne(TimeSpan.FromMilliseconds(100)));
+    }
+
+    public async Task DisposeAsync()
+    {
+        using var stopTimeout = new CancellationTokenSource(WaitTimeout);
+        await _service.StopAsync(stopTimeout.Token);
+        _service.Dispose();
+        _probe.Dispose();
+        _producer.Dispose();
+    }
+
+    [Theory]
+    [InlineData("""{"taskId":"other-task","workflowInstanceId":"workflow-1","status":"CANCELED","workflowTask":null}""")]
+    [InlineData("not json")]
+    public async Task CancelsTaskAfterUnprocessableMessage(string unprocessableMessage)
+    {
+        using var runningTask = _notifier.GetCancellationToken("running-task", CancellationToken.None);
+
+        await _producer.ProduceAsync(_partition, new Message<Null, string> { Value = unprocessableMessage });
+        await _producer.ProduceAsync(_partition, StatusEvent("running-task", "CANCELED"));
+
+        Assert.True(runningTask.CancellationToken.WaitHandle.WaitOne(WaitTimeout), "The running task was not cancelled");
+    }
+
+    [Fact]
+    public async Task StopAsyncEndsConsumption()
+    {
+        using var stopTimeout = new CancellationTokenSource(WaitTimeout);
+        await _service.StopAsync(stopTimeout.Token);
+
+        Assert.True(_service.ExecuteTask!.IsCompleted, "The consumer is still running after StopAsync");
+    }
+
+    private static Message<Null, string> StatusEvent(string taskId, string status) =>
+        new()
+        {
+            Value = JsonConvert.SerializeObject(
+                new
+                {
+                    taskId,
+                    workflowInstanceId = "workflow-1",
+                    status,
+                    workflowTask = new
+                    {
+                        name = TaskName,
+                        taskReferenceName = "task_ref",
+                        type = "SIMPLE"
+                    }
+                }
+            )
+        };
+}