diff --git a/Obvs.Kafka.Tests/Hosting/TestKafkaHosting.cs b/Obvs.Kafka.Tests/Hosting/TestKafkaHosting.cs index cad2971..77b5e92 100644 --- a/Obvs.Kafka.Tests/Hosting/TestKafkaHosting.cs +++ b/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(); + Action handle = service => service.AsServer() + .HandleCommand((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().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(); var busB = hostB.Services.GetRequiredService(); - var receivedByA = busA.Commands.OfType().FirstAsync().ToTask(); - var receivedByB = busB.Commands.OfType().FirstAsync().ToTask(); - await busA.SendAsync(new PlaceOrder { OrderId = "o-1" }); + var receivedByA = busA.Events.OfType().FirstAsync().ToTask(); + var receivedByB = busB.Events.OfType().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(TaskCreationOptions.RunContinuationsAsynchronously); + Action handle = service => service.AsServer() + .HandleCommand((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().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(TaskCreationOptions.RunContinuationsAsynchronously); + Action handle = service => service.AsClient() + .HandleEvent((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().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 StartHostAsync(string serviceName, Action configure = null) + private static async Task StartHostAsync(string serviceName, Action configure = null, string instanceId = null) { var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true }); - builder.Services.AddObvs(obvs => obvs.AddService(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(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 WaitForAsync(Func 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. + } } } diff --git a/Obvs.Kafka.Tests/TestKafkaConsumerGroups.cs b/Obvs.Kafka.Tests/TestKafkaConsumerGroups.cs new file mode 100644 index 0000000..2ce6182 --- /dev/null +++ b/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("orders.Commands").ConsumerConfig.GroupId); + Assert.Equal("orders.Requests", provider.CreateSource("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("orders.Events").ConsumerConfig.GroupId); + Assert.Equal("orders.Responses.bus-a", provider.CreateSource("orders.Responses").ConsumerConfig.GroupId); + Assert.Equal("orders.Events.bus-b", Provider("bus-b").CreateSource("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(topic).ConsumerConfig; + Assert.Equal(config.ClientId, config.GroupId); + } + } + + private static KafkaServiceEndpointProvider Provider(string instanceId) + { + return new KafkaServiceEndpointProvider("orders", + new KafkaConfiguration("kafka.example:9092"), null, null, + new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)), + null, null, null, null, instanceId); + } + } +} diff --git a/Obvs.Kafka.Tests/TestMessageSourceErrors.cs b/Obvs.Kafka.Tests/TestMessageSourceErrors.cs new file mode 100644 index 0000000..b1a9fb9 --- /dev/null +++ b/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 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 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(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(configuration, new KafkaSourceConfiguration(), topic, + new IMessageDeserializer[] { new JsonMessageDeserializer() }, null); + var publisher = new MessagePublisher(configuration, new KafkaProducerConfiguration(), topic, new JsonMessageSerializer(), null); + var received = new ConcurrentQueue(); + Exception error = null; + + using (source.Messages.OfType().Subscribe(received.Enqueue, exception => error = exception)) + using (var producer = new ProducerBuilder(new ProducerConfig { BootstrapServers = TestBroker.BootstrapServers }).Build()) + { + // Not an Obvs message, so the consumer can't deserialize it + await producer.ProduceAsync(topic, new Message { 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 Source(ILogger logger) + { + var configuration = new KafkaConfiguration("kafka.example:9093") { Logger = logger }; + return new MessageSource(configuration, new KafkaSourceConfiguration(), "orders.Commands", new IMessageDeserializer[0], null); + } + + private sealed class RecordingObserver : IObserver + { + 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 state) => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func formatter) + { + Entries.Add((logLevel, formatter(state, exception))); + } + } + } +} diff --git a/Obvs.Kafka/Configuration/KafkaConfiguration.cs b/Obvs.Kafka/Configuration/KafkaConfiguration.cs index ae1f782..15ef893 100644 --- a/Obvs.Kafka/Configuration/KafkaConfiguration.cs +++ b/Obvs.Kafka/Configuration/KafkaConfiguration.cs @@ -2,6 +2,8 @@ using System.Linq; using System.Net; using Confluent.Kafka; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; namespace Obvs.Kafka.Configuration { @@ -38,6 +40,9 @@ internal KafkaConfiguration(KafkaOptions options) // Set on each client's builder when signing in with OAUTHBEARER internal Action OAuthBearerTokenRefreshHandler { get; } + // For errors the Kafka clients recover from by themselves, which don't end a subscription + internal ILogger Logger { get; set; } = NullLogger.Instance; + // Sets the security settings on each client's config internal void ApplyTo(ClientConfig config) { diff --git a/Obvs.Kafka/Configuration/KafkaServiceEndpointProvider.cs b/Obvs.Kafka/Configuration/KafkaServiceEndpointProvider.cs index 8eb5080..907734d 100644 --- a/Obvs.Kafka/Configuration/KafkaServiceEndpointProvider.cs +++ b/Obvs.Kafka/Configuration/KafkaServiceEndpointProvider.cs @@ -23,9 +23,13 @@ public class KafkaServiceEndpointProvider _typeFilter; private readonly Func, bool> _propertyFiter; private readonly Func> _propertyProvider; - private readonly string _uniqueGroupSuffix; + private readonly string _instanceId; - /// Gives each source its own consumer group, so several buses in one process each receive every message + /// + /// The bus's instance ID. With one, the service's buses share a consumer group for commands and requests, + /// so each is handled once, and each bus has its own for events and responses, named after its instance ID. + /// Without one, as in v6, every source has the process's group, so every process receives every message. + /// public KafkaServiceEndpointProvider(string serviceName, KafkaConfiguration kafkaConfiguration, KafkaSourceConfiguration sourceConfiguration, @@ -33,7 +37,7 @@ public KafkaServiceEndpointProvider(string serviceName, IMessageSerializer serializer, IMessageDeserializerFactory deserializerFactory, Func, bool> propertyFilter, Func> propertyProvider, - Func assemblyFilter = null, Func typeFilter = null, string uniqueGroupSuffix = null) + Func assemblyFilter = null, Func typeFilter = null, string instanceId = null) : base(serviceName) { _kafkaConfiguration = kafkaConfiguration; @@ -45,7 +49,7 @@ public KafkaServiceEndpointProvider(string serviceName, _typeFilter = typeFilter; _propertyFiter = propertyFilter; _propertyProvider = propertyProvider; - _uniqueGroupSuffix = uniqueGroupSuffix; + _instanceId = instanceId; if (string.IsNullOrEmpty(_kafkaConfiguration?.BootstrapServers)) { @@ -56,17 +60,28 @@ public KafkaServiceEndpointProvider(string serviceName, public override IServiceEndpoint CreateEndpoint() { return new ServiceEndpoint( - CreateSource(_kafkaConfiguration, _sourceConfiguration, RequestsDestination), - CreateSource(_kafkaConfiguration, _sourceConfiguration, CommandsDestination), + CreateSource(RequestsDestination), + CreateSource(CommandsDestination), Create(_kafkaConfiguration, _producerConfiguration, EventsDestination), Create(_kafkaConfiguration, _producerConfiguration, ResponsesDestination), typeof(TServiceMessage)); } - private IMessageSource CreateSource(KafkaConfiguration kafkaConfiguration, KafkaSourceConfiguration sourceConfiguration, string topic) where T : class, TMessage + internal MessageSource CreateSource(string topic) where T : class, TMessage { - return SourceFactory.Create(kafkaConfiguration, sourceConfiguration, - topic, _deserializerFactory, _propertyFiter, _assemblyFilter, _typeFilter, _uniqueGroupSuffix); + return SourceFactory.Create(_kafkaConfiguration, _sourceConfiguration, + topic, _deserializerFactory, _propertyFiter, _assemblyFilter, _typeFilter, GroupId(topic)); + } + + // Named after the service, and not the process, so a restarted bus reuses its groups. Each topic has its own + // group, so one source's subscription doesn't rebalance the others. + private string GroupId(string topic) + { + if (string.IsNullOrEmpty(_instanceId)) + { + return null; + } + return topic == CommandsDestination || topic == RequestsDestination ? topic : $"{topic}.{_instanceId}"; } private IMessagePublisher Create(KafkaConfiguration kafkaConfiguration, KafkaProducerConfiguration producerConfiguration, string topic) where T : class, TMessage @@ -77,8 +92,8 @@ private IMessagePublisher Create(KafkaConfiguration kafkaConfiguration, Ka public override IServiceEndpointClient CreateEndpointClient() { return new ServiceEndpointClient( - CreateSource(_kafkaConfiguration, _sourceConfiguration, EventsDestination), - CreateSource(_kafkaConfiguration, _sourceConfiguration, ResponsesDestination), + CreateSource(EventsDestination), + CreateSource(ResponsesDestination), Create(_kafkaConfiguration, _producerConfiguration, RequestsDestination), Create(_kafkaConfiguration, _producerConfiguration, CommandsDestination), typeof(TServiceMessage)); diff --git a/Obvs.Kafka/Configuration/SourceFactory.cs b/Obvs.Kafka/Configuration/SourceFactory.cs index 4f12b79..aaf448b 100644 --- a/Obvs.Kafka/Configuration/SourceFactory.cs +++ b/Obvs.Kafka/Configuration/SourceFactory.cs @@ -15,7 +15,7 @@ public static MessageSource Create( Func, bool> propertyFilter, Func assemblyFilter = null, Func typeFilter = null, - string uniqueGroupSuffix = null) + string groupId = null) where TMessage : class where TServiceMessage : class { @@ -25,7 +25,7 @@ public static MessageSource Create( topic, deserializerFactory.Create(assemblyFilter, typeFilter), propertyFilter, - uniqueGroupSuffix); + groupId); } } } \ No newline at end of file diff --git a/Obvs.Kafka/KafkaTransportExtensions.cs b/Obvs.Kafka/KafkaTransportExtensions.cs index ef9aa58..e02e537 100644 --- a/Obvs.Kafka/KafkaTransportExtensions.cs +++ b/Obvs.Kafka/KafkaTransportExtensions.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Obvs.Configuration; using Obvs.Kafka; @@ -89,7 +90,8 @@ private static ObvsServiceBuilder UseKafka(ObvsServiceBuilder service, Func Validate(service, GetOptions(serviceProvider, service), whereToSet)) - .UseEndpointProvider(serviceProvider => CreateProvider(service, GetOptions(serviceProvider, service))); + .UseEndpointProvider(serviceProvider => CreateProvider(service, GetOptions(serviceProvider, service), + serviceProvider.GetService()?.CreateLogger("Obvs.Kafka"))); } private static KafkaOptions GetOptions(IServiceProvider serviceProvider, ObvsServiceBuilder service) @@ -114,9 +116,13 @@ private static IEnumerable Validate(ObvsServiceBuilder service, KafkaOpt } // The provider is generic in the service's message type, which the builder only knows at run time - private static IServiceEndpointProvider CreateProvider(ObvsServiceBuilder service, KafkaOptions options) + private static IServiceEndpointProvider CreateProvider(ObvsServiceBuilder service, KafkaOptions options, ILogger logger) { var kafkaConfiguration = new KafkaConfiguration(options); + if (logger != null) + { + kafkaConfiguration.Logger = logger; + } KafkaTopics.EnsureCreated(service.Name, kafkaConfiguration, new[] { $"{service.Name}.Commands", $"{service.Name}.Events", $"{service.Name}.Requests", $"{service.Name}.Responses", diff --git a/Obvs.Kafka/MessageSource.cs b/Obvs.Kafka/MessageSource.cs index 641270e..9c59f62 100644 --- a/Obvs.Kafka/MessageSource.cs +++ b/Obvs.Kafka/MessageSource.cs @@ -10,6 +10,7 @@ using System.Threading; using System.Threading.Tasks; using Confluent.Kafka; +using Microsoft.Extensions.Logging; using Obvs.Hosting; using Obvs.Kafka.Configuration; using Obvs.Serialization; @@ -27,18 +28,19 @@ public class MessageSource : IMessageSource private readonly ConsumerConfig _consumerConfig; private readonly Action _oauthBearerTokenRefreshHandler; + private readonly ILogger _logger; // How long Subscribe waits for Kafka to assign partitions, and for each offset query while it does private static readonly TimeSpan PartitionAssignmentTimeout = TimeSpan.FromSeconds(30); private static readonly TimeSpan OffsetQueryTimeout = TimeSpan.FromSeconds(10); - /// Only required if you intend to create two or more message sources in the same process + /// The consumer group. Without one, the group is named after the process, so each process receives every message public MessageSource(KafkaConfiguration kafkaConfig, KafkaSourceConfiguration sourceConfig, string topicName, IEnumerable> deserializers, Func, bool> propertyFilter, - string uniqueGroupSuffix = null) + string groupId = null) { _deserializers = deserializers.ToDictionary(d => d.GetTypeName()); _topicName = topicName; @@ -47,8 +49,7 @@ public MessageSource(KafkaConfiguration kafkaConfig, _consumerConfig = new ConsumerConfig { ClientId = clientId, - // A group of its own, so it receives every message and other sources' subscriptions don't rebalance it - GroupId = string.IsNullOrEmpty(uniqueGroupSuffix) ? clientId : $"{clientId}-{topicName}-{uniqueGroupSuffix}", + GroupId = string.IsNullOrEmpty(groupId) ? clientId : groupId, BootstrapServers = kafkaConfig.BootstrapServers, FetchMinBytes = sourceConfig.FetchMinBytes, FetchMaxBytes = sourceConfig.FetchMaxBytes, @@ -59,6 +60,7 @@ public MessageSource(KafkaConfiguration kafkaConfig, }; kafkaConfig.ApplyTo(_consumerConfig); _oauthBearerTokenRefreshHandler = kafkaConfig.OAuthBearerTokenRefreshHandler; + _logger = kafkaConfig.Logger; } internal ConsumerConfig ConsumerConfig => _consumerConfig; @@ -94,39 +96,49 @@ private void Subscriber(CancellationToken token, IObserver observer, M { try { - var builder = new ConsumerBuilder(_consumerConfig) - .SetValueDeserializer(new ConsumerValueDeserializer()) - .SetErrorHandler((_, e) => observer.OnError(new Exception($"Consumer ErrorHandler: {e.Reason}"))) - .SetPartitionsAssignedHandler((c, partitions) => - { - var offsets = StartOffsets(c, partitions); - subscribedEvent.Set(); - return offsets; - }); - if (_oauthBearerTokenRefreshHandler != null) - { - builder.SetOAuthBearerTokenRefreshHandler(_oauthBearerTokenRefreshHandler); - } - - using (var consumer = builder.Build()) + // Cancelled when the subscription ends, or by the error handler, which runs on this thread as a side + // effect of Consume, when an error ends it. Consume then stops waiting for a message. + using (var consuming = CancellationTokenSource.CreateLinkedTokenSource(token)) { - consumer.Subscribe(_topicName); - - try - { - while (!token.IsCancellationRequested) + var builder = new ConsumerBuilder(_consumerConfig) + .SetValueDeserializer(new ConsumerValueDeserializer()) + .SetErrorHandler((_, error) => { - ConsumeMessage(token, observer, consumer); - } - } - catch (Exception exception) + if (OnClientError(error, observer)) + { + consuming.Cancel(); + } + }) + .SetPartitionsAssignedHandler((c, partitions) => + { + var offsets = StartOffsets(c, partitions, _logger); + subscribedEvent.Set(); + return offsets; + }); + if (_oauthBearerTokenRefreshHandler != null) { - observer.OnError(exception); + builder.SetOAuthBearerTokenRefreshHandler(_oauthBearerTokenRefreshHandler); } - finally + + using (var consumer = builder.Build()) { - consumer.Unsubscribe(); - consumer.Close(); + consumer.Subscribe(_topicName); + + try + { + while (!consuming.IsCancellationRequested && ConsumeMessage(consuming.Token, observer, consumer)) + { + } + } + catch (Exception exception) + { + observer.OnError(exception); + } + finally + { + consumer.Unsubscribe(); + consumer.Close(); + } } } } @@ -145,16 +157,32 @@ private void Subscriber(CancellationToken token, IObserver observer, M // Fix each partition's start position at assignment: the group's committed offset, or else the current // end of the log. Leaving "latest" to be resolved later would skip messages published in the meantime. - private static IEnumerable StartOffsets(IConsumer consumer, List partitions) + // The end of the log is committed straight away, so a bus that takes the partition over, or this one + // restarted, carries on from there even if nothing was consumed, rather than skipping to the end again. + private static IEnumerable StartOffsets(IConsumer consumer, List partitions, ILogger logger) { - return consumer.Committed(partitions, OffsetQueryTimeout) - .Select(committed => committed.Offset != Offset.Unset - ? committed - : new TopicPartitionOffset(committed.TopicPartition, consumer.QueryWatermarkOffsets(committed.TopicPartition, OffsetQueryTimeout).High)) + var committed = consumer.Committed(partitions, OffsetQueryTimeout); + var fromEnd = committed + .Where(offset => offset.Offset == Offset.Unset) + .Select(offset => new TopicPartitionOffset(offset.TopicPartition, consumer.QueryWatermarkOffsets(offset.TopicPartition, OffsetQueryTimeout).High)) .ToList(); + if (fromEnd.Count > 0) + { + try + { + consumer.Commit(fromEnd); + } + catch (KafkaException exception) + { + // This consumer still starts from there; only a later one might not + logger.LogWarning(exception, "Kafka consumer couldn't commit its start position: {Reason}", exception.Error.Reason); + } + } + return committed.Where(offset => offset.Offset != Offset.Unset).Concat(fromEnd).ToList(); } - private void ConsumeMessage(CancellationToken token, IObserver observer, IConsumer consumer) + // Returns false once the subscription has ended + private bool ConsumeMessage(CancellationToken token, IObserver observer, IConsumer consumer) { try { @@ -168,11 +196,43 @@ private void ConsumeMessage(CancellationToken token, IObserver observe SetHeaders(message, kafkaMessage.Headers); observer.OnNext(message); } + return true; + } + catch (ConsumeException exception) + { + // Including a message that isn't an Obvs message, which has been consumed and is skipped + return !OnClientError(exception, observer); } catch (Exception exception) { observer.OnError(exception); + return false; + } + } + + // Returns true when the error ends the subscription + internal bool OnClientError(Error error, IObserver observer) + { + return OnClientError(new KafkaException(error), observer); + } + + // librdkafka recovers by itself from the errors it reports, such as brokers that are down or a failed OAUTHBEARER + // token refresh, unless they're fatal. Being refused access to a topic, the group or the cluster doesn't fix + // itself either. Only those end the subscription; the rest are logged and consumption carries on. + private bool OnClientError(KafkaException exception, IObserver observer) + { + var error = exception.Error; + if (error.IsFatal + || error.Code == ErrorCode.TopicAuthorizationFailed + || error.Code == ErrorCode.GroupAuthorizationFailed + || error.Code == ErrorCode.ClusterAuthorizationFailed) + { + observer.OnError(exception); + return true; } + + _logger.LogWarning(exception, "Kafka consumer of {Topic} reported an error, and carries on: {Reason}", _topicName, error.Reason); + return false; } private TMessage DeserializePayload(WrapperMessage msg) diff --git a/docs/transports/kafka-amazon-msk.md b/docs/transports/kafka-amazon-msk.md index 0c16ee7..867deaf 100644 --- a/docs/transports/kafka-amazon-msk.md +++ b/docs/transports/kafka-amazon-msk.md @@ -53,7 +53,7 @@ The role or user needs an IAM policy allowing, on the cluster, its topics and it - `kafka-cluster:Connect` - `kafka-cluster:DescribeTopic`, `kafka-cluster:CreateTopic` (or create the topics beforehand) - `kafka-cluster:WriteData`, `kafka-cluster:ReadData` -- `kafka-cluster:DescribeGroup`, `kafka-cluster:AlterGroup` on `group//*`. Each bus names its own consumer groups after the machine and process, so allow any group name. +- `kafka-cluster:DescribeGroup`, `kafka-cluster:AlterGroup` on `group//orders.*`, which covers all the service's [consumer groups](kafka.md#consumer-groups). The Java client's `AWS_MSK_IAM` mechanism isn't available in .NET. `OAUTHBEARER` with the signer is the .NET equivalent. @@ -81,7 +81,7 @@ MSK keeps SCRAM credentials in AWS Secrets Manager, in a secret whose name start service.UseKafka(builder.Configuration.GetSection("Obvs:Orders:Kafka")); ``` -Read the password from Secrets Manager into configuration, for example with a configuration provider, rather than putting it in `appsettings.json`. With SCRAM, MSK authorises with Kafka ACLs, so the user needs `CREATE` (or topics created beforehand), `READ` and `WRITE` on the topics, and `READ` on any consumer group. +Read the password from Secrets Manager into configuration, for example with a configuration provider, rather than putting it in `appsettings.json`. With SCRAM, MSK authorises with Kafka ACLs, so the user needs `CREATE` (or topics created beforehand), `READ` and `WRITE` on the topics, and `READ` on consumer groups with the prefix `orders.`. ## TLS diff --git a/docs/transports/kafka-azure-event-hubs.md b/docs/transports/kafka-azure-event-hubs.md index 4e7a38f..de4e132 100644 --- a/docs/transports/kafka-azure-event-hubs.md +++ b/docs/transports/kafka-azure-event-hubs.md @@ -75,4 +75,8 @@ Give the identity **Azure Event Hubs Data Sender** and **Azure Event Hubs Data R ## Consumer groups -Each bus uses consumer groups of its own, named after the machine and process, so every restart uses new ones. Event Hubs allows 1,000 Kafka consumer groups per namespace, and groups left behind by stopped buses can count towards it until they're removed, so a service that restarts often, or runs many instances, can approach the limit. +Event Hubs allows **1,000 Kafka consumer groups per namespace**, in each tier with a Kafka endpoint. Kafka consumer groups are created as they're used, and don't need creating in the portal like Event Hubs' own consumer groups. + +Each service uses two [consumer groups](kafka.md#consumer-groups) shared by its buses, for commands and requests, and two for each bus, for events and responses, named after the bus's instance ID. That's a new GUID each time a bus starts, unless it's set with `UseInstanceId(...)`, so by default every restart adds two groups, and the old ones stay until they're removed. Microsoft doesn't document when Event Hubs removes unused Kafka consumer groups, so a service that restarts often, or runs many instances, can approach the limit. + +To stay well within it, give each bus a fixed instance ID of its own with `UseInstanceId(...)`, such as the pod name in a Kubernetes StatefulSet, so a restarted bus reuses its groups. It then also receives the events published while it was stopped. diff --git a/docs/transports/kafka-confluent-cloud.md b/docs/transports/kafka-confluent-cloud.md index 6bee824..206513b 100644 --- a/docs/transports/kafka-confluent-cloud.md +++ b/docs/transports/kafka-confluent-cloud.md @@ -73,7 +73,7 @@ At startup the bus creates its four topics (`orders.Commands`, `orders.Events`, - `CREATE` on the topics, or create them beforehand; - `READ` and `WRITE` on the topics; -- `READ` on consumer groups with any name (`*`), because each bus names its own consumer groups after the machine and process. +- `READ` on consumer groups with the prefix `orders.`, which all the service's [consumer groups](kafka.md#consumer-groups) start with. ## Startup checks diff --git a/docs/transports/kafka.md b/docs/transports/kafka.md index 73fe9df..56bdc9c 100644 --- a/docs/transports/kafka.md +++ b/docs/transports/kafka.md @@ -72,9 +72,51 @@ The bus checks the settings when it starts, and reports every problem together i When the brokers refuse to sign the bus in, or the TLS handshake fails, startup fails straight away with the broker's reason. It doesn't wait out the connection timeout. +## Consumer groups + +Each topic is read in a consumer group of its own, named after the topic and not after the machine or process, so a restarted bus reuses its groups and carries on where they left off: + +| Messages | Consumer group | Received by | +|---|---|---| +| Commands | `.Commands`, shared by the service's buses | One of the service's buses | +| Requests | `.Requests`, shared by the service's buses | One of the service's buses | +| Events | `.Events.`, one per bus | Every bus | +| Responses | `.Responses.`, one per bus | Every bus, which keeps the responses to its own requests | + +- Kafka gives each partition of a topic to one member of a group, so commands and requests are spread across the service's buses only as far as their topics have partitions. With one partition, one bus handles them all, and another takes over if it stops. Topics created at startup get the broker's default number of partitions; to spread the work over more buses, create the topics beforehand with more. +- A group that's new starts at the end of its topic. After that, a bus carries on from the group's committed position, so commands and requests sent while none of the service's buses was running are handled when one starts. +- Positions are committed every few seconds as messages are received, not after they're handled. A bus that crashes can receive the last few messages again, or lose one it was handling. + +The instance ID is the bus's `InstanceId`, a new GUID each time the bus starts unless it's set with `UseInstanceId(...)`. With a new one, a bus starts with new events and responses groups, at the end of the topics, so it doesn't receive events published while it was stopped. To keep its groups across restarts, give each bus a fixed instance ID of its own, such as the pod name in a Kubernetes StatefulSet: + +```csharp +builder.Services.AddObvs(obvs => obvs + .UseInstanceId(Environment.GetEnvironmentVariable("POD_NAME")) + .AddService("orders", service => service + .UseKafka(builder.Configuration.GetSection("Obvs:Orders:Kafka")) + .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>))))); +``` + +Two running buses mustn't share an instance ID: Kafka would split the events between them. + +### Groups left behind + +A bus without a fixed instance ID leaves its events and responses groups behind when it stops. Kafka removes a group once it's empty and its committed positions have expired, `offsets.retention.minutes` after its last member left: seven days by default. Until then the groups count towards any limit on consumer groups, such as [Event Hubs'](kafka-azure-event-hubs.md#consumer-groups). To remove them sooner, delete them with Kafka's `kafka-consumer-groups.sh --bootstrap-server --delete --group `. + +Services configured with v6's `WithKafkaEndpoints<...>()` keep v6's consumer groups, named after the machine and process, so every process receives every message, commands included. + +## Errors while running + +The Kafka client recovers by itself from most errors, such as brokers that are down, connections the platform closed, or a failed OAUTHBEARER token refresh, where it keeps the token it has and tries again. The bus logs those as warnings, in the `Obvs.Kafka` category, and carries on. A message the Kafka client can't deserialize, such as one another producer wrote to the topic in its own format, is logged and skipped. + +These end the subscription instead, with the client's error: + +- errors the client reports as fatal (`Error.IsFatal`), after which it stops consuming; +- being refused access to a topic, a consumer group or the cluster, which won't fix itself. + ## Permissions -At startup the bus creates the service's topics if they don't exist. That needs permission to create topics; without it, create them beforehand. Each bus also uses consumer groups of its own, named after the machine and process, so on platforms with access control, grant read access to consumer groups by any name, not one fixed name. +At startup the bus creates the service's topics if they don't exist. That needs permission to create topics; without it, create them beforehand. The service's consumer groups are all named after it, starting with `.`, so on platforms with access control, grant read access to consumer groups with that prefix. ## Trying it locally