From 12922f77bc240ffdbb8fad9c6067716e92217db3 Mon Sep 17 00:00:00 2001 From: "milan.svitlica" Date: Fri, 9 Oct 2026 09:35:53 +0200 Subject: [PATCH] Keep the Kafka cancellation consumer running after a bad message (3.10.1 backport) 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.Engine.Tests.csproj | 1 + .../KafkaCancellationNotifierTests.cs | 105 ++++++++++++++++++ 8 files changed, 124 insertions(+), 17 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 6a50ebad..1531681a 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 3.10.0 + 3.10.1 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 c8feb708..b0444c01 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 3.10.0 + 3.10.1 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 c18731e0..fa9561f2 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 3.10.0 + 3.10.1 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 ecb51225..1a8265a1 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 3.10.0 + 3.10.1 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" + } + } + ) + }; +}