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" + } + } + ) + }; +}