Obvs.Kafka: carry on after transient errors, and name consumer groups after the service

Obvs.Kafka.Tests/Hosting/TestKafkaHosting.cs

@@ -7,6 +7,7 @@
 using System.Reactive.Threading.Tasks;
 using System.Threading.Tasks;
 using Confluent.Kafka;
+using Confluent.Kafka.Admin;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Obvs.Hosting;
@@ -70,23 +71,98 @@ public async Task ShouldReturnResponsesOverKafka()
         }
 
         [Fact, Trait("Category", "Explicit")]
-        public async Task ShouldDeliverToEveryBusInTheProcess()
+        public async Task ShouldHandleEachCommandOnceAcrossTheServicesBuses()
         {
-            // As with the other transports, each bus receives every message, so they mustn't share a consumer group
+            // The service's buses share a consumer group for commands, so Kafka gives each command to one of them
+            var serviceName = _serviceNames.Create("command-once");
+            var handled = new ConcurrentQueue<string>();
+            Action<ObvsServiceBuilder> handle = service => service.AsServer()
+                .HandleCommand<PlaceOrder>((command, context, ct) =>
+                {
+                    handled.Enqueue(command.OrderId);
+                    return Task.CompletedTask;
+                });
+            using var hostA = await StartHostAsync(serviceName, handle);
+            using var hostB = await StartHostAsync(serviceName, handle);
+            using var sender = await StartHostAsync(serviceName, service => service.AsClient());
+            var orderIds = Enumerable.Range(0, 10).Select(i => $"o-{i}").ToList();
+
+            foreach (var orderId in orderIds)
+            {
+                await sender.Services.GetRequiredService<IServiceBusClient>().SendAsync(new PlaceOrder { OrderId = orderId });
+            }
+
+            await WaitForAsync(() => handled.Count >= orderIds.Count ? handled : null);
+            // Long enough for a second delivery to arrive, if there were one
+            await Task.Delay(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken);
+            Assert.Equal(orderIds, handled.OrderBy(orderId => orderId));
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldDeliverEveryEventToEveryBus()
+        {
+            // As with the other transports, each bus receives every event, so they have consumer groups of their own
             var serviceName = _serviceNames.Create("every-bus");
             using var hostA = await StartHostAsync(serviceName);
             using var hostB = await StartHostAsync(serviceName);
             var busA = hostA.Services.GetRequiredService<IServiceBus>();
             var busB = hostB.Services.GetRequiredService<IServiceBus>();
 
-            var receivedByA = busA.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
-            var receivedByB = busB.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
-            await busA.SendAsync(new PlaceOrder { OrderId = "o-1" });
+            var receivedByA = busA.Events.OfType<OrderPlaced>().FirstAsync().ToTask();
+            var receivedByB = busB.Events.OfType<OrderPlaced>().FirstAsync().ToTask();
+            await busA.PublishAsync(new OrderPlaced { OrderId = "o-1" });
 
             Assert.Equal("o-1", (await receivedByA.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
             Assert.Equal("o-1", (await receivedByB.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
         }
 
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldHandleCommandsSentWhileTheServiceWasDown()
+        {
+            // The commands group outlives the bus, so a bus started later, even with another instance ID, carries on from it
+            var serviceName = _serviceNames.Create("commands-while-down");
+            var handled = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
+            Action<ObvsServiceBuilder> handle = service => service.AsServer()
+                .HandleCommand<PlaceOrder>((command, context, ct) =>
+                {
+                    handled.TrySetResult(command.OrderId);
+                    return Task.CompletedTask;
+                });
+            var first = await StartHostAsync(serviceName, handle);
+            await first.StopAsync(TestContext.Current.CancellationToken);
+            first.Dispose();
+            using var sender = await StartHostAsync(serviceName, service => service.AsClient());
+
+            await sender.Services.GetRequiredService<IServiceBusClient>().SendAsync(new PlaceOrder { OrderId = "o-1" });
+            using var restarted = await StartHostAsync(serviceName, handle);
+
+            Assert.Equal("o-1", await handled.Task.WaitAsync(Timeout, TestContext.Current.CancellationToken));
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldReceiveEventsPublishedWhileABusWithAFixedInstanceIdWasDown()
+        {
+            // The bus's events group is named after its instance ID, so restarting with the same one carries on from it
+            var serviceName = _serviceNames.Create("events-while-down");
+            var instanceId = $"subscriber-{Guid.NewGuid():N}";
+            var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
+            Action<ObvsServiceBuilder> handle = service => service.AsClient()
+                .HandleEvent<OrderPlaced>((ev, context, ct) =>
+                {
+                    received.TrySetResult(ev.OrderId);
+                    return Task.CompletedTask;
+                });
+            var first = await StartHostAsync(serviceName, handle, instanceId);
+            await first.StopAsync(TestContext.Current.CancellationToken);
+            first.Dispose();
+            using var publisher = await StartHostAsync(serviceName, service => service.AsServer());
+
+            await publisher.Services.GetRequiredService<IServiceBus>().PublishAsync(new OrderPlaced { OrderId = "o-1" });
+            using var restarted = await StartHostAsync(serviceName, handle, instanceId);
+
+            Assert.Equal("o-1", await received.Task.WaitAsync(Timeout, TestContext.Current.CancellationToken));
+        }
+
         [Fact, Trait("Category", "Explicit")]
         public async Task ShouldKeepWorkingWhenAnotherHostInTheProcessStops()
         {
@@ -158,16 +234,23 @@ public async Task ShouldCarryHeadersTheAppSetsOverKafka()
             Assert.Equal("acme", headers["x-tenant"]);
         }
 
-        private static async Task<IHost> StartHostAsync(string serviceName, Action<ObvsServiceBuilder> configure = null)
+        private static async Task<IHost> StartHostAsync(string serviceName, Action<ObvsServiceBuilder> configure = null, string instanceId = null)
         {
             var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true });
-            builder.Services.AddObvs(obvs => obvs.AddService<IOrderMessage>(serviceName, service =>
+            builder.Services.AddObvs(obvs =>
             {
-                service
-                    .UseKafka(options => options.BootstrapServers = TestBroker.BootstrapServers)
-                    .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)));
-                configure?.Invoke(service);
-            }));
+                if (instanceId != null)
+                {
+                    obvs.UseInstanceId(instanceId);
+                }
+                obvs.AddService<IOrderMessage>(serviceName, service =>
+                {
+                    service
+                        .UseKafka(options => options.BootstrapServers = TestBroker.BootstrapServers)
+                        .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)));
+                    configure?.Invoke(service);
+                });
+            });
             var host = builder.Build();
             await host.StartAsync(TestContext.Current.CancellationToken);
             return host;
@@ -184,7 +267,7 @@ private static async Task<T> WaitForAsync<T>(Func<T> find) where T : class
             return found ?? throw new TimeoutException($"The expected {typeof(T).Name} wasn't recorded in time.");
         }
 
-        // Names services uniquely for this run, and deletes their topics afterwards
+        // Names services uniquely for this run, and deletes their topics and consumer groups afterwards
         public sealed class ServiceNames : IAsyncDisposable
         {
             private readonly string _runId = Guid.NewGuid().ToString("N").Substring(0, 8);
@@ -194,11 +277,24 @@ public sealed class ServiceNames : IAsyncDisposable
             public async ValueTask DisposeAsync()
             {
                 using var admin = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = TestBroker.BootstrapServers }).Build();
-                var topics = admin.GetMetadata(Timeout).Topics.Select(topic => topic.Topic).Where(topic => topic.StartsWith("obvs-kafka-") && topic.Contains(_runId)).ToList();
+                bool IsThisRuns(string name) => name.StartsWith("obvs-kafka-") && name.Contains(_runId);
+                var topics = admin.GetMetadata(Timeout).Topics.Select(topic => topic.Topic).Where(IsThisRuns).ToList();
                 if (topics.Count > 0)
                 {
                     await admin.DeleteTopicsAsync(topics);
                 }
+                var groups = (await admin.ListConsumerGroupsAsync()).Valid.Select(group => group.GroupId).Where(IsThisRuns).ToList();
+                try
+                {
+                    if (groups.Count > 0)
+                    {
+                        await admin.DeleteGroupsAsync(groups);
+                    }
+                }
+                catch (DeleteGroupsException exception) when (exception.Results.All(result => result.Error.Code == ErrorCode.NoError || result.Error.Code == ErrorCode.NonEmptyGroup))
+                {
+                    // Groups of hosts that were disposed without being stopped still have members. Kafka removes them later.
+                }
             }
         }
 

Obvs.Kafka.Tests/TestKafkaConsumerGroups.cs

@@ -0,0 +1,59 @@
+using Obvs.Kafka.Configuration;
+using Obvs.Serialization.Json;
+using Obvs.Types;
+using Xunit;
+
+namespace Obvs.Kafka.Tests
+{
+    // How consumer groups are named. Names don't depend on the machine or process, so a restarted bus reuses its groups.
+    // None of these connect to a broker.
+    public class TestKafkaConsumerGroups
+    {
+        public interface ITestMessage : IMessage
+        {
+        }
+
+        [Fact]
+        public void ShouldShareOneGroupPerServiceForCommandsAndRequests()
+        {
+            // So each command and request is handled by one of the service's buses
+            foreach (var provider in new[] { Provider("bus-a"), Provider("bus-b") })
+            {
+                Assert.Equal("orders.Commands", provider.CreateSource<ICommand>("orders.Commands").ConsumerConfig.GroupId);
+                Assert.Equal("orders.Requests", provider.CreateSource<IRequest>("orders.Requests").ConsumerConfig.GroupId);
+            }
+        }
+
+        [Fact]
+        public void ShouldGiveEachBusItsOwnGroupForEventsAndResponses()
+        {
+            // So every bus receives every event, and the responses to its own requests
+            var provider = Provider("bus-a");
+
+            Assert.Equal("orders.Events.bus-a", provider.CreateSource<IEvent>("orders.Events").ConsumerConfig.GroupId);
+            Assert.Equal("orders.Responses.bus-a", provider.CreateSource<IResponse>("orders.Responses").ConsumerConfig.GroupId);
+            Assert.Equal("orders.Events.bus-b", Provider("bus-b").CreateSource<IEvent>("orders.Events").ConsumerConfig.GroupId);
+        }
+
+        [Fact]
+        public void ShouldKeepV6sGroupNamedAfterTheProcessWithoutAnInstanceId()
+        {
+            // v6's fluent configuration, where every source in the process receives every message
+            var provider = Provider(null);
+
+            foreach (var topic in new[] { "orders.Commands", "orders.Events", "orders.Requests", "orders.Responses" })
+            {
+                var config = provider.CreateSource<IMessage>(topic).ConsumerConfig;
+                Assert.Equal(config.ClientId, config.GroupId);
+            }
+        }
+
+        private static KafkaServiceEndpointProvider<ITestMessage, IMessage, ICommand, IEvent, IRequest, IResponse> Provider(string instanceId)
+        {
+            return new KafkaServiceEndpointProvider<ITestMessage, IMessage, ICommand, IEvent, IRequest, IResponse>("orders",
+                new KafkaConfiguration("kafka.example:9092"), null, null,
+                new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)),
+                null, null, null, null, instanceId);
+        }
+    }
+}

Obvs.Kafka.Tests/TestMessageSourceErrors.cs

@@ -0,0 +1,156 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
+using Confluent.Kafka;
+using Confluent.Kafka.Admin;
+using Microsoft.Extensions.Logging;
+using Obvs.Kafka.Configuration;
+using Obvs.Serialization;
+using Obvs.Serialization.Json;
+using Xunit;
+using IMessage = Obvs.Types.IMessage;
+
+namespace Obvs.Kafka.Tests
+{
+    // Which errors the Kafka client reports end a subscription. The client recovers from most of them by itself.
+    public class TestMessageSourceErrors
+    {
+        public interface ITestMessage : IMessage
+        {
+        }
+
+        public class TestMessage : ITestMessage
+        {
+            public string Data { get; set; }
+        }
+
+        public static IEnumerable<object[]> ErrorsTheClientRecoversFrom => new[]
+        {
+            new object[] { new Error(ErrorCode.Local_AllBrokersDown, "1/1 brokers are down") },
+            new object[] { new Error(ErrorCode.Local_Transport, "kafka.example:9093/0: Disconnected (after 240000ms in state UP)") },
+            new object[] { new Error(ErrorCode.Local_Resolve, "Failed to resolve 'kafka.example:9093'") },
+            new object[] { new Error(ErrorCode.Local_TimedOut, "Timed out") },
+            // An OAUTHBEARER token refresh that failed: librdkafka keeps the token it has and tries again in 10 seconds
+            new object[] { new Error(ErrorCode.Local_Authentication, "Failed to acquire SASL OAUTHBEARER token: token endpoint unavailable") },
+            new object[] { new Error(ErrorCode.Local_ValueDeserialization, "Invalid wire type") },
+            new object[] { new Error(ErrorCode.UnknownTopicOrPart, "Subscribed topic not available: orders.Commands") },
+        };
+
+        public static IEnumerable<object[]> ErrorsThatEndTheSubscription => new[]
+        {
+            new object[] { new Error(ErrorCode.Local_Fatal, "Fatal error: fenced by another instance", true) },
+            new object[] { new Error(ErrorCode.FencedInstanceId, "Fenced by another static member", true) },
+            new object[] { new Error(ErrorCode.TopicAuthorizationFailed, "Not authorized to access topics: [orders.Commands]") },
+            new object[] { new Error(ErrorCode.GroupAuthorizationFailed, "Not authorized to access group: orders.Commands") },
+            new object[] { new Error(ErrorCode.ClusterAuthorizationFailed, "Cluster authorization failed") },
+        };
+
+        [Theory]
+        [MemberData(nameof(ErrorsTheClientRecoversFrom))]
+        public void ShouldLogAndCarryOnAfterAnErrorTheClientRecoversFrom(Error error)
+        {
+            var logger = new RecordingLogger();
+            var source = Source(logger);
+            var observer = new RecordingObserver();
+
+            var ends = source.OnClientError(error, observer);
+
+            Assert.False(ends);
+            Assert.Null(observer.Error);
+            var entry = Assert.Single(logger.Entries);
+            Assert.Equal(LogLevel.Warning, entry.Level);
+            Assert.Contains(error.Reason, entry.Message);
+            Assert.Contains("orders.Commands", entry.Message);
+        }
+
+        [Theory]
+        [MemberData(nameof(ErrorsThatEndTheSubscription))]
+        public void ShouldEndTheSubscriptionOnAnErrorThatWontFixItself(Error error)
+        {
+            var source = Source(new RecordingLogger());
+            var observer = new RecordingObserver();
+
+            var ends = source.OnClientError(error, observer);
+
+            Assert.True(ends);
+            var exception = Assert.IsType<KafkaException>(observer.Error);
+            Assert.Equal(error.Code, exception.Error.Code);
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldSkipAMessageItCantReadAndCarryOn()
+        {
+            var topic = $"obvs-kafka-unreadable-{Guid.NewGuid():N}";
+            using var admin = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = TestBroker.BootstrapServers }).Build();
+            await admin.CreateTopicsAsync(new[] { new TopicSpecification { Name = topic, NumPartitions = 1, ReplicationFactor = 1 } });
+            try
+            {
+                var configuration = new KafkaConfiguration(TestBroker.BootstrapServers);
+                var source = new MessageSource<ITestMessage>(configuration, new KafkaSourceConfiguration(), topic,
+                    new IMessageDeserializer<ITestMessage>[] { new JsonMessageDeserializer<TestMessage>() }, null);
+                var publisher = new MessagePublisher<ITestMessage>(configuration, new KafkaProducerConfiguration(), topic, new JsonMessageSerializer(), null);
+                var received = new ConcurrentQueue<TestMessage>();
+                Exception error = null;
+
+                using (source.Messages.OfType<TestMessage>().Subscribe(received.Enqueue, exception => error = exception))
+                using (var producer = new ProducerBuilder<Null, byte[]>(new ProducerConfig { BootstrapServers = TestBroker.BootstrapServers }).Build())
+                {
+                    // Not an Obvs message, so the consumer can't deserialize it
+                    await producer.ProduceAsync(topic, new Message<Null, byte[]> { Value = new byte[] { 0xFF, 0xFF, 0xFF, 0xFF } }, TestContext.Current.CancellationToken);
+                    await publisher.PublishAsync(new TestMessage { Data = "after the unreadable one" });
+
+                    var deadline = DateTime.UtcNow.AddSeconds(10);
+                    while (received.IsEmpty && error == null && DateTime.UtcNow < deadline)
+                    {
+                        await Task.Delay(100, TestContext.Current.CancellationToken);
+                    }
+                }
+
+                Assert.Null(error);
+                Assert.Equal("after the unreadable one", Assert.Single(received).Data);
+            }
+            finally
+            {
+                await admin.DeleteTopicsAsync(new[] { topic });
+            }
+        }
+
+        private static MessageSource<ITestMessage> Source(ILogger logger)
+        {
+            var configuration = new KafkaConfiguration("kafka.example:9093") { Logger = logger };
+            return new MessageSource<ITestMessage>(configuration, new KafkaSourceConfiguration(), "orders.Commands", new IMessageDeserializer<ITestMessage>[0], null);
+        }
+
+        private sealed class RecordingObserver : IObserver<ITestMessage>
+        {
+            public Exception Error { get; private set; }
+
+            public void OnCompleted()
+            {
+            }
+
+            public void OnError(Exception error) => Error = error;
+
+            public void OnNext(ITestMessage value)
+            {
+            }
+        }
+
+        private sealed class RecordingLogger : ILogger
+        {
+            public List<(LogLevel Level, string Message)> Entries { get; } = new List<(LogLevel, string)>();
+
+            public IDisposable BeginScope<TState>(TState state) => null;
+
+            public bool IsEnabled(LogLevel logLevel) => true;
+
+            public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func<TState, Exception, string> formatter)
+            {
+                Entries.Add((logLevel, formatter(state, exception)));
+            }
+        }
+    }
+}