From 101d90377bcb0588600bc27c64ec06abb2d2b2c1 Mon Sep 17 00:00:00 2001 From: "milan.svitlica" Date: Fri, 9 Oct 2026 09:34:20 +0200 Subject: [PATCH] Keep the Kafka cancellation consumer running after a bad message (4.4.2) The consume loop sat inside a single try/catch. When one message failed, the catch logged "will retry to consume after 5 seconds", waited, and then ExecuteAsync returned. The consumer never restarted, so the worker kept running but stopped cancelling tasks until the process restarted. Two kinds of message trigger it: - the handler throws, e.g. a status event with "workflowTask": null (NullReferenceException on WorkflowTask.Type); - the value fails to deserialize (invalid or empty JSON, an unknown status, a workflowTask without name or taskReferenceName), so Consume throws a ConsumeException. Errors are now caught per message: logged, and the loop moves on to the next message. OperationCanceledException from the stopping token still ends the loop, so shutdown is unchanged. The 5-second pause is gone. It only ran once before the loop exited, and pausing on every bad message would hold up every cancellation behind it. Tests run the real consumer against librdkafka's in-process mock cluster (test.mock.num.brokers), so they need no broker or Docker. Co-Authored-By: Claude Opus 5.5 --- .../ConductorSharp.Client.csproj | 2 +- .../ConductorSharp.Engine.csproj | 2 +- .../AssemblyInfo.cs | 3 + ...ctorSharp.KafkaCancellationNotifier.csproj | 2 +- .../Service/KafkaConsumerBackgroundService.cs | 24 ++-- .../ConductorSharp.Patterns.csproj | 2 +- .../ConductorSharp.Toolkit.csproj | 2 +- .../ConductorSharp.Engine.Tests.csproj | 1 + .../KafkaCancellationNotifierTests.cs | 105 ++++++++++++++++++ 9 files changed, 125 insertions(+), 18 deletions(-) create mode 100644 src/ConductorSharp.KafkaCancellationNotifier/AssemblyInfo.cs create mode 100644 test/ConductorSharp.Engine.Tests/Integration/KafkaCancellationNotifierTests.cs diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 1a506640..b88a9d53 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 4.4.1 + 4.4.2 Client library for Netflix Conductor, with some additional quality of life features. https://github.com/codaxy/conductor-sharp netflix;conductor diff --git a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index df478cee..97235c2e 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 4.4.1 + 4.4.2 Client library for Netflix Conductor, with some additional quality of life features. https://github.com/codaxy/conductor-sharp netflix;conductor diff --git a/src/ConductorSharp.KafkaCancellationNotifier/AssemblyInfo.cs b/src/ConductorSharp.KafkaCancellationNotifier/AssemblyInfo.cs new file mode 100644 index 00000000..f7591f45 --- /dev/null +++ b/src/ConductorSharp.KafkaCancellationNotifier/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("ConductorSharp.Engine.Tests")] diff --git a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj index 989e052e..7595db51 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 4.4.1 + 4.4.2 Codaxy Codaxy diff --git a/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaConsumerBackgroundService.cs b/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaConsumerBackgroundService.cs index 02677733..cb4d7af3 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaConsumerBackgroundService.cs +++ b/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaConsumerBackgroundService.cs @@ -35,7 +35,6 @@ public TaskStatusModel Deserialize(ReadOnlySpan data, bool isNull, Seriali private readonly IOptions _kafkaOptions; private readonly KafkaCancellationNotifier _notifier; private readonly ILogger _logger; - private const int KafkaRetryPeriodSeconds = 5; public KafkaConsumerBackgroundService( IOptions 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 ); diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index 70163e9e..c7b264b2 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 4.4.1 + 4.4.2 diff --git a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj index 0b48430b..37157bae 100644 --- a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj +++ b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj @@ -7,7 +7,7 @@ disable true dotnet-conductorsharp - 4.4.1 + 4.4.2 diff --git a/test/ConductorSharp.Engine.Tests/ConductorSharp.Engine.Tests.csproj b/test/ConductorSharp.Engine.Tests/ConductorSharp.Engine.Tests.csproj index a67cc2a7..be3912b5 100644 --- a/test/ConductorSharp.Engine.Tests/ConductorSharp.Engine.Tests.csproj +++ b/test/ConductorSharp.Engine.Tests/ConductorSharp.Engine.Tests.csproj @@ -94,6 +94,7 @@ + diff --git a/test/ConductorSharp.Engine.Tests/Integration/KafkaCancellationNotifierTests.cs b/test/ConductorSharp.Engine.Tests/Integration/KafkaCancellationNotifierTests.cs new file mode 100644 index 00000000..92dfaeaf --- /dev/null +++ b/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 _producer = new ProducerBuilder( + new Dictionary { ["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.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.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 { 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 StatusEvent(string taskId, string status) => + new() + { + Value = JsonConvert.SerializeObject( + new + { + taskId, + workflowInstanceId = "workflow-1", + status, + workflowTask = new + { + name = TaskName, + taskReferenceName = "task_ref", + type = "SIMPLE" + } + } + ) + }; +}