From 3be32c888b8eddbcc31659af8ced4ba598280c8e Mon Sep 17 00:00:00 2001 From: Rafael Lillo <7280959+lillo42@users.noreply.github.com> Date: Thu, 23 Jul 2026 20:48:36 +0100 Subject: [PATCH 1/3] Document Kafka KIP-848 consumer group protocol (BrighterCommand/Brighter#4233) - KafkaConfiguration: add Consumer Group Protocol (KIP-848) section covering IGroupProtocol, ClassicGroupProtocol, ConsumerGroupProtocol, back-fill behavior, no-sharing contract, static membership and Validate caveats - KafkaConfiguration: mark SessionTimeout and PartitionAssignmentStrategy subscription options as deprecated in favor of GroupProtocol - V10MigrationGuide: add Kafka KIP-848 as a V10 new feature with before/after migration examples for the obsoleted properties --- contents/KafkaConfiguration.md | 479 +------------------- contents/V10MigrationGuide.md | 801 +-------------------------------- 2 files changed, 2 insertions(+), 1278 deletions(-) diff --git a/contents/KafkaConfiguration.md b/contents/KafkaConfiguration.md index 02211d3..8694137 100644 --- a/contents/KafkaConfiguration.md +++ b/contents/KafkaConfiguration.md @@ -1,478 +1 @@ -# Kafka Configuration - -## General - -Kafka is OSS message-oriented-middleware and is [well documented](https://kafka.apache.org/documentation/#gettingStarted). Brighter handles the details of sending to or receiving from Kafka. You may find it useful to understand the [building blocks](https://kafka.apache.org/documentation/#introduction) of the protocol. Brighter's Kafka support is implemented on top of the Confluent .NET client, and you might find the [documentation for the .NET client](https://docs.confluent.io/kafka-clients/dotnet/current/overview.html) helpful when debugging, but you should not have to interact with it directly to use Brighter (although we expose many of its configuration options). - -Kafka has two main roles: - -- **Producer**: A producer sends events to a **Topic** on a Kafka broker. -- **Consumer**: A consumer reads events from a **Topic** on a Kafka broker. - -**Topics** are append-only streams of events. Multiple producers can write to a topic, and multiple consumers can read from one. A **consumer** uses an **offset** into the stream to indicate the event it wants to read. Kafka does not delete an event from the stream when it is ack'd by the consumer; instead a **consumer** increments its **offset** once an item has been read so that it can avoid processing the same event twice. See [Offset Management](#offset-management) for more on how Brighter manages **consumer offsets**. As a result the lifetime of events on a stream is instead a configuration setting for the stream. - -As a **consumer** manages an **offset** to record events that is has read, you cannot scale an application that wishes to consume a **topic** by increasing the number of **consumers**--they don't share an offset--without partitioning the **topic**. If you supply a **partition key**, a **partition** uses consistent hashing to slice a **topic** into a number of streams; otherwise it will use round-robin. See [this documentation](https://jaceklaskowski.gitbooks.io/apache-kafka/content/kafka-producer-internals-DefaultPartitioner.html) for more. Each **partition** is only read by a single **consumer** within the application. All of the consumers for an application should share the same group id, called a **consumer group** in Kafka. As each **consumer** tracks the **offset** for the **partitions** it is reading, it is possible to have multiple **consumers** read and process the same **topic**. - -A **consumer** may read from *multiple* **partitions**, but only one **consumer** may read from a **partition** at one time in a given **consumer group**. Kafka will assign partitions across the pool of consumers for the **consumer group**. When the pool changes, a **rebalance** occurs, which may mean that a consumer changes the **partition** that it is assigned within the **consumer group**. Brighter favors *sticky assignment of partitions* to avoid unnecessary churn of partitions. - -In addition to the Producer API and Consumer API Kafka streams have features such as the Streams API and the Connect API. We do not use either of these from Brighter. - -## Connection - -The Connection to Kafka is provided by an **KafkaMessagingGatewayConnection** which allows you to configure the following: - -- **BootstrapServers**: A **bootstrap** server is a well-known broker through which we discover the servers in the Kafka cluster that we can connect to. You should supply a comma-separated list of host and port pairs. These are the addresses of the Kafka brokers in the "bootstrap" Kafka cluster. -- **Debug**: A comma-separated list of debug contexts to enable. Producer: broker, topic, msg. Consumer: consumer, cgrp, topic, fetch. -- **Name**: An identifier to use for the client. -- **SaslMechanisms**: If any, what is the protocol used for authenticated connection to the Kafka broker: plain, scram-sha-256, scram-sha-256, gssapi (kerberos), oauthbearer -- **SaslKerberosName**: If using kerberos, what is the connection name. -- **SaslUsername**: SASL username for use with PLAIN and SASL-SCRAM -- **SaslPassword**: SASL password for use with PLAIN and SASL-SCRAM -- **SecurityProtocol**: How are messages between client and server encrypted, if at all: plaintext, ssl, saslplaintext, saslssl -- **SslCaLocation**: Where is the CA certificate located (see [here](https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/csharp.html) for guidance). -- **SslKeystoreLocation**: Path to the client's keystore -- **SslKeystorePassword**: Password for the client's keystore - -The following code connects to a local Kafka instance (for development): - -``` csharp - services.AddBrighter(...) - .AddProducers((configure) => - { - configure.ProducerRegistry = new KafkaProducerRegistryFactory( - new KafkaMessagingGatewayConfiguration() - { - Name = "paramore.brighter.greetingsender", - BootStrapServers = new[] {"localhost:9092"} - }, - ...//publication, see below - ) - .Create(); - }) - ... - -``` - -The following code connects to a remote Kafka instance. The settings here will depend on how your production broker is configured for access. We show getting secrets from environment variables for simplicity, again you will need to adjust this for your approach to secrets management: - -``` csharp - services.AddBrighter(...) - .AddProducers((configure) => - { - configure.ProducerRegistry = new KafkaProducerRegistryFactory( - new KafkaMessagingGatewayConfiguration() - { - Name = "paramore.brighter.greetingsender", - BootStrapServers = new[] { Environment.GetEnvironmentVariable("BOOSTRAP_SERVER")}, - SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol.SaslSsl, - SaslMechanisms = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism.Plain, - SaslUsername = Environment.GetEnvironmentVariable("SASL_USERNAME"), - SaslPassword = Environment.GetEnvironmentVariable("SASL_PASSWORD"), - SslCaLocation = RuntimeInformation.IsOSPlatform(OSPlatform.OSX) ? "/usr/local/etc/openssl@1.1/cert.pem" : null; - }, - ...//publication, see below - ) - .Create(); - }) - ... - -``` - -## Publication - -For more on a *Publication* see the material on an *Add Producers* in [Basic Configuration](/contents/BrighterBasicConfiguration.md#using-an-external-bus). - -We allow you to configure properties for both Brighter and the Confluent .NET client. Because there are many properties on the Confluent .NET Client we also configure a callback to let you inspect and modify the configuration that we will pass to the client if you so desire. This can be used to add properties we do not support or adjust how we set them. - -- **Replication**: how many ISR nodes must receive the record before the producer can consider the write successful. Default is Acks.All. -- **BatchNumberMessages**: Maximum number of messages batched in one MessageSet. Default is 10. -- **EnableIdempotence**: Messages are produced once only. Will adjust the following if not set: `max.in.flight.requests.per.connection=5` (must be less than or equal to 5), `retries=INT32_MAX` (must be greater than 0), `acks=all`, `queuing.strategy=fifo`. Default is true. -- **LingerMs**: Maximum time, in milliseconds, for buffering data on the producer queue. Default is 5. -- **MessageSendMaxRetries**: How many times to retry sending a failing MessageSet. Note: retrying may cause reordering, set the max in flight to 1 if you need ordering by when sent. Default is 3. -- **MessageTimeoutMs**: Local message timeout. This value is only enforced locally and limits the time a produced message waits for successful delivery. A time of 0 is infinite. Default is 5000. -- **MaxInFlightRequestsPerConnection**: Maximum number of in-flight requests the client will send. We default this to 1, so as to allow retries to not de-order the stream. -- **NumPartitions**: How many partitions for this topic. We default to 1. -- **Partitioner**: How do we partition? Defaults to Partitioner.ConsistentRandom. -- **QueueBufferingMaxMessages**: Maximum number of messages allowed on the producer queue. Defaults to 10. -- **QueueBufferingMaxKbytes**: Maximum total message size sum allowed on the producer queue. Defaults to 1048576 bytes (so for 10 messages about 104Kb per message). -- **ReplicationFactor**: What is the replication factor? How many nodes is the topic copied to on the broker? Defaults to 1. -- **RetryBackoff**: The backoff time before retrying a message send. Defaults to 100. -- **RequestTimeoutMs**: The ack timeout of the producer request. This value is only enforced by the broker and relies on Replication being != AcksEnum.None. Defaults to 500. -- **TopicFindTimeoutMs**: How long to wait when asking for topic metadata. Defaults to 5000. -- **TransactionalId**: The unique identifier for this producer, used with transactions - -The following example shows how a *Publication* might be configured: - -``` csharp - services.AddBrighter(...) - .AddProducers((configure) => - { - configure.ProducerRegistry = new KafkaProducerRegistryFactory( - ...,//connection see above - new KafkaPublication[] {new KafkaPublication() - { - Topic = new RoutingKey("MyTopicName"), - NumPartitions = 3, - ReplicationFactor = 3, - MessageTimeoutMs = 1000, - RequestTimeoutMs = 1000, - MakeChannels = OnMissingChannel.Create - }} - ).Create(); - }) - -``` - -### Configuration Callback - -The Confluent .NET client has a range of configuration options. Some of those can be controlled through the publication. But, to allow you the full range of configuration options for the Confluent client, including new options that may appear, we provide a callback on the **KafkaProducerRegistryFactory**. The registry exposes a method, **SetConfigHook(Action hook)**. The method takes a *delegate* (you can pass a lambda). Your delegate will be called with the *proposed* ProducerConfig (taking into account the *Publication* settings). You can adjust additional parameters at this point. - -You can use it as follows: - -``` csharp - - var publication = new KafkaPublication() - { - Topic = new RoutingKey("MyTopicName"), - NumPartitions = 3, - ReplicationFactor = 3, - MessageTimeoutMs = 1000, - RequestTimeoutMs = 1000, - MakeChannels = OnMissingChannel.Create - }; - publication.SetConfigHook(config => config.EnableGaplessGuarantee = true) - - services.AddBrighter(...) - .AddProducers((configure) => - { - configure.ProducerRegistry = new KafkaProducerRegistryFactory( - ...,//connection see above - new KafkaPublication() {publication} - ).Create(); - }) - -``` - -### Kafka Topic Auto Create - -Brighter uses the Kafka AdminClient for topic creation. For this to work as expected you should set the server property of **auto.create.topics.enable** to **false**; otherwise the topic will be auto-created with the values defined by your server for new topics, such as the number of partitions. This error can be insidious because your code will still work against this topic, but without inspection you will not observe that its properties do not match those requested. - -If you want to specify the topic through Brighter, or through your own IaaS code, we recommend always setting this setting to false; we recommend only setting it to true if you tell Brighter to assume that the infrastructure exists, as it will then be created on the first write. - -## Subscription - -For more on a *Subscription* see the material on configuring *Service Activator* in [Basic Configuration](/contents/BrighterBasicConfiguration.md#configuring-the-service-activator). - -We support a number of Kafka specific *Subscription* options: - -- **CommitBatchSize**: We commit processed work (marked as acked or rejected) when a batch size worth of work has been completed (see [below](#offset-management)). Defaults to 10. -- **ConfigHook**: Allows you to modify the Kafka client configuration before a consumer is created. Used to set properties that Brighter does not expose. See [Configuration Callback](#configuration-callback) below. -- **GroupId**: Only one consumer in a group can read from a partition at any one time; this preserves ordering. We do not default this value, and expect you to set it. -- **IsolationLevel**: Default to read only committed messages, change if you want to read uncommitted messages. May cause duplicates. Defaults to ReadCommitted. -- **MaxPollInterval**: How often the consumer needs to poll for new messages to be considered alive, polling greater than this interval triggers a re-balance. Defaults to 300000ms (5 minutes). -- **NumPartitions**: How many partitions does the topic have? Used for topic creation, if required. Defaults to 1. -- **OffsetDefault**: What do we do if there is no offset stored for this consumer. Defaults to AutoOffsetReset.Earliest - Begin reading the stream from the start. Options include AutoOffsetReset.Latest - Start from now i.e. only consume messages after we start and AutoOffsetReset.Error - which considers it an error if no reset is found -- **PartitionAssignmentStrategy**: How do partitions get assigned to consumers in the group? Defaults to RoundRobin for even distribution. See [Partition Assignment Strategy](#partition-assignment-strategy) below. -- **ReadCommittedOffsetsTimeOut**: How long before attempting to read back committed offsets (mainly used in debugging) is an error. Defaults to 5000ms. -- **ReplicationFactor**: What is the replication factor? How many nodes is the topic copied to on the broker? Defaults to 1. Used for topic creation if required. -- **SessionTimeout**: If Kafka does not receive a heartbeat from the consumer within this time window, trigger a re-balance. Defaults to 10000ms (10 seconds). -- **SweepUncommittedOffsetsInterval**: The interval at which we sweep, looking for offsets that have not been flushed (see [below](#offset-management)). Defaults to 30000ms (30 seconds). - -The following example shows how a subscription might be configured: - -``` csharp - private static void ConfigureBrighter(HostBuilderContext hostContext, IServiceCollection services) -{ - var subscriptions = new KafkaSubscription[] - { - new KafkaSubscription( - subscriptionName: new SubscriptionName("paramore.example.greeting"), - channelName: new ChannelName("greeting.event"), - routingKey: new RoutingKey("greeting.event"), - groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID"), - timeOut: TimeSpan.FromMilliseconds(100), - commitBatchSize: 5, - sweepUncommittedOffsetsInterval: TimeSpan.FromMilliseconds(3000), - messagePumpType: MessagePumpType.Reactor - ) - }; - - //create the gateway - var consumerFactory = new KafkaMessageConsumerFactory( - new KafkaMessagingGatewayConfiguration {...} // see connection information above - ); - - services.AddConsumers(options => - { - options.Subscriptions = subscriptions; - options.ChannelFactory = new ChannelFactory(consumerFactory); - }).AutoFromAssemblies(); - - - services.AddHostedService(); -} -``` - -### Configuration Callback - -Similar to producers, the Confluent .NET client for consumers has a range of configuration options. Some of those can be controlled through the subscription. But, to allow you the full range of configuration options for the Confluent client, including new options that may appear, we provide a **configHook** parameter on **KafkaSubscription**. - -The **configHook** takes a *delegate* (you can pass a lambda). Your delegate will be called with the *proposed* ConsumerConfig (taking into account the *Subscription* settings). You can adjust additional parameters at this point. - -You can use it as follows: - -``` csharp -var subscription = new KafkaSubscription( - subscriptionName: new SubscriptionName("paramore.example.greeting"), - channelName: new ChannelName("greeting.event"), - routingKey: new RoutingKey("greeting.event"), - groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID"), - timeOut: TimeSpan.FromMilliseconds(100), - commitBatchSize: 5, - sweepUncommittedOffsetsInterval: TimeSpan.FromMilliseconds(3000), - messagePumpType: MessagePumpType.Reactor, - configHook: config => - { - // Customize the Confluent Consumer configuration - config.FetchMinBytes = 1024; - config.FetchWaitMaxMs = 500; - config.StatisticsIntervalMs = 60000; // Enable statistics - } -); - -var subscriptions = new KafkaSubscription[] { subscription }; - -//create the gateway -var consumerFactory = new KafkaMessageConsumerFactory( - new KafkaMessagingGatewayConfiguration {...} // see connection information above -); - -services.AddConsumers(options => -{ - options.Subscriptions = subscriptions; - options.ChannelFactory = new ChannelFactory(consumerFactory); -}).AutoFromAssemblies(); -``` - -### Common Configuration Callback Use Cases - -The **configHook** is useful for: - -**1. Fine-tuning fetch behavior**: - -``` csharp -configHook: config => -{ - config.FetchMinBytes = 1024; // Minimum data per fetch request - config.FetchWaitMaxMs = 100; // Max time to wait for fetch.min.bytes - config.FetchMaxBytes = 52428800; // Maximum data for all partitions -} -``` - -**2. Enabling consumer statistics**: - -``` csharp -configHook: config => -{ - config.StatisticsIntervalMs = 30000; // Emit stats every 30 seconds -} -``` - -**3. Customizing security settings**: - -``` csharp -configHook: config => -{ - config.EnableSslCertificateVerification = true; - config.SslEndpointIdentificationAlgorithm = SslEndpointIdentificationAlgorithm.Https; -} -``` - -**4. Adjusting partition assignment**: - -``` csharp -configHook: config => -{ - // Note: PartitionAssignmentStrategy should typically be set via the subscription parameter - // But you can override it here if needed - config.SessionTimeoutMs = 45000; // Increase for slow rebalances -} -``` - -**5. Debugging and monitoring**: - -``` csharp -configHook: config => -{ - config.Debug = "consumer,cgrp,topic,fetch"; // Enable debug logging - config.StatisticsIntervalMs = 10000; // Frequent statistics for debugging -} -``` - -### Partition Assignment Strategy - -Kafka distributes partitions across consumers in a consumer group using a partition assignment strategy. In V10, Brighter provides control over this strategy through the **partitionAssignmentStrategy** parameter. - -**Available Strategies**: - -- **RoundRobin** (default): Distributes partitions evenly across consumers in a round-robin fashion. This is the most balanced distribution but may cause all partitions to be reassigned during rebalances. -- **Range**: Assigns contiguous partition ranges to consumers. Useful for co-locating related partitions on the same consumer. -- **CooperativeSticky**: Not supported with Brighter's manual offset commits (will throw an exception). - -**Why RoundRobin is the Default**: - -Brighter uses **RoundRobin** as the default strategy because: -1. It provides the most even distribution of partitions across consumers -2. It works reliably with Brighter's manual offset commit strategy -3. It's straightforward to reason about for most use cases - -**Using Range Strategy**: - -The **Range** strategy is useful when you want to co-locate partitions on the same consumer: - -``` csharp -var subscription = new KafkaSubscription( - subscriptionName: new SubscriptionName("paramore.example.greeting"), - channelName: new ChannelName("greeting.event"), - routingKey: new RoutingKey("greeting.event"), - groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID"), - partitionAssignmentStrategy: PartitionAssignmentStrategy.Range -); -``` - -**CooperativeSticky Limitation**: - -The **CooperativeSticky** strategy is not supported when using manual offset commits (which Brighter requires for its at-least-once delivery guarantees). Attempting to use it will throw an `ArgumentOutOfRangeException`: - -``` csharp -// This will throw an exception: -var subscription = new KafkaSubscription( - // ... - partitionAssignmentStrategy: PartitionAssignmentStrategy.CooperativeSticky // ❌ Not supported -); -``` - -This is due to a [known issue in librdkafka](https://github.com/confluentinc/librdkafka/issues/4059) where CooperativeSticky doesn't work correctly with manual offset management. - -## Offset Management - -It is important to understand how Brighter manages the **offset** of any **partitions** assigned to your **consumer**. - -- Brighter manages committing **offsets** to Kafka. This means we set the Confluent client's *auto store* and *auto commit* properties to *false*. -- The **CommitBatchSize** setting on the *Subscription* determines the size of your buffer. A smaller buffer is less efficient, but if your consumer crashes any **offsets** pending commit in the buffer will be lost, and you will be represented with those records when you next read from the **partition**. We default this value to 10. -- We do not add an offset commit to the buffer until you Ack the request. The message pump will Ack for you once you exit your handler (via return or [throwing an exception](/contents/HandlerFailure.md)). -- Flushing the commit buffer happens on a separate thread. We only run one flush at a time, and we flush a **CommitBatchSize** number of items from the buffer. - - A busy consumer may not flush on every increment of the **CommitBatchSize**, as it may need to wait for the last flush to finish. - - We won't flush again until we cross the next multiple of the **CommitBatchSize**. For example if the **CommitBatchSize** is 10, and the handler is busy so that by the time the buffer flushes there are 13 pending commits in the buffer, the buffer would only flush 10, and 3 would remain in the buffer; we would not flush the next 10 until the buffer hit 20. - - If your **CommitBatchSize** is too low for the throughput, you might find that you miss a flush interval, because you are already flushing. - - If you miss a flush on a busy consumer, your buffer will begin to back up. If this continues, you will not catch up with subsequent flushes, which only flush the **CommitBatchSize** each time. This would lead to you continually being "backed up". - - For this reason you must set a **CommitBatchSize** that keeps pace with the throughput of your consumer. Use a larger **CommitBatchSize** for higher throughput consumers, smaller for lower. -- We sweep uncommitted offsets at an interval. This triggers a flush if no flush has run since the last flush plus the *Subscription's* **SweepUncommittedOffsetsIntervalMs**. - - A sweep will not run if a flush is currently running (and will in turn block a flush). - - A sweep flushes a **CommitBatchSize** worth of commits. - - It is intended for low-throughput consumers where commits might otherwise languish waiting for a batch-size increment. - - It is *not* intended to flush a buffer that backs up because the **CommitBatchSize** is too low, and won't function for that. Fix the **CommitBatchSize** instead. -- On a re-balance where we stop processing a **partition** on an individual consumer, we flush the remaining **offsets** for the revoked **partitions**. - - We configure the consumer to use sticky assignment strategy to avoid unnecessary re-assignments (see the [Confluent documentation](https://www.confluent.io/blog/cooperative-rebalancing-in-kafka-streams-consumer-ksqldb/)). -- On a consumer shutdown we flush the buffer to commit all **offsets**. - -## Working with Schema Registry - -If you want to use tools within the Kafka ecosystem such as Kafka Connect or KQSL you will almost certainly need to use Confluent Schema Registry to provide the schema of your message. - -You will need to pull in the following package: - -* Confluent.SchemaRegistry - -and a package for the serialization of your choice. Here we are using JSON, so we use - -* Confluent.SchemaRegistry.Serdes.Json - -When working with Brighter, to use Confluent Schema Registry you will need to take a dependency on ISchemaRegistry in the constructor of your message mapper. To fulfill this constructor, in your application setup you will need to register an instance of schema registry. You should configure the schema registry config url to be the url of you schema registry. (Here we just use localhost for a development instance running in docker as an example). - -``` csharp -var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081"}; -var cachedSchemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryConfig); -services.AddSingleton(cachedSchemaRegistryClient); -``` - -Once you can satisfy the dependency, you will want to use the serializer from the Serdes package to serialize the body of your message, instead of System.Text.Json. Note that 'under-the-hood' the Serdes serializer uses [Json.NET](https://www.newtonsoft.com/json) and [NJsonSchema](https://github.com/RicoSuter/NJsonSchema), so you may need to mark up your code with attributes from these packages to create the schema you want and serialize a valid message to it. (Note that, at this time, the Serdes package does not support System.Text.Json so you will need to take a dependency on Json.NET if you want to use the schema registry). - -It is worth noting the following aspects of the code sample below: - -* We need to set up a SerializationContext and tell Serdes that we are serializing the message body using their serializer -* We provide two helpers, though you can pass your own settings if you prefer: - * **ConfluentJsonSerializationConfig.SerdesJsonSerializerConfig()** offers default settings for JSON serialization (many of these are passed through to Json.NET). - * **ConfluentJsonSerializationConfig.NJsonSchemaGeneratorSettings()** offers default settings for JSON Schema generation (such as using camelCase). - -``` csharp -public class GreetingEventMessageMapper : IAmAMessageMapper -{ - private readonly ISchemaRegistryClient _schemaRegistryClient; - private readonly string _partitionKey = "KafkaTestQueueExample_Partition_One"; - private SerializationContext _serializationContext; - private const string Topic = "greeting.event"; - - public GreetingEventMessageMapper(ISchemaRegistryClient schemaRegistryClient) - { - _schemaRegistryClient = schemaRegistryClient; - //We care about ensuring that we serialize the body using the Confluent tooling, as it registers and validates schema - _serializationContext = new SerializationContext(MessageComponentType.Value, Topic); - } - - public Message MapToMessage(GreetingEvent request) - { - var header = new MessageHeader(messageId: request.Id, topic: Topic, messageType: MessageType.MT_EVENT); - //This uses the Confluent JSON serializer, which wraps Newtonsoft but also performs schema registration and validation - var serializer = new JsonSerializer(_schemaRegistryClient, ConfluentJsonSerializationConfig.SerdesJsonSerializerConfig(), ConfluentJsonSerializationConfig.NJsonSchemaGeneratorSettings()).AsSyncOverAsync(); - var s = serializer.Serialize(request, _serializationContext); - var body = new MessageBody(s, "JSON"); - header.PartitionKey = _partitionKey; - - var message = new Message(header, body); - return message; - } - - public GreetingEvent MapToRequest(Message message) - { - var deserializer = new JsonDeserializer().AsSyncOverAsync(); - //This uses the Confluent JSON serializer, which wraps Newtonsoft but also performs schema registration and validation - var greetingCommand = deserializer.Deserialize(message.Body.Bytes, message.Body.Bytes is null, _serializationContext); - - return greetingCommand; - } -} - -``` - -## Requeue with Delay (Non-Blocking Retry) - -We don't currently support requeue with delay for Kafka. This is also known as non-blocking retry. With a stream if your app cannot process a record but it might be able to process the record after a delay (for example the DB is temporarily unavailable) then the options are: - -* Blocking Retry - keep retrying the processing of this record -* Load Shedding - ack the record to commit the offset, skipping this record -* Non-Blocking Retry - move the record to a new store or queue, skipping the original, append after a delay - -Brighter supports the first two of these options. - -* Blocking Retry - use a Polly policy via the **UsePolicy** attribute -* Load Shedding - allow the handler to complete, or throw an exception. This will cause the handler to commit the offset. - -Note that Blocking Retry means you will apply backpressure as the blocking retry means you will pause consumption until the record can be processed. - -A non-blocking retry typically creates a copy of the current record, and appends it to the stream so that it can be processed later: - -- Publish the message to be requeued to a new stream or store with a timestamp -- Ack the existing message so at to commit the offset -- Poll that stream or store and publish anything whose timestamp + delay means it is now due - -(You may need multiple tables or streams to support different delay lengths) - -Until Brighter supports this for you, implementation of non-blocking consumers is left to the user. - - - - - - - +__KAFKA__ \ No newline at end of file diff --git a/contents/V10MigrationGuide.md b/contents/V10MigrationGuide.md index 7cc802b..5c3d92d 100644 --- a/contents/V10MigrationGuide.md +++ b/contents/V10MigrationGuide.md @@ -1,800 +1 @@ -# Brighter V10 Migration Guide - -## Overview - -Brighter V10 introduces significant improvements and new features while maintaining a clear migration path from V9. This guide provides step-by-step instructions for upgrading your application to V10, addressing breaking changes, and adopting new features. - -**Key Changes in V10**: - -- Cloud Events support -- OpenTelemetry Semantic Conventions -- Default Message Mappers -- Dynamic Message Deserialization -- Nullable Reference Types (breaking) -- Simplified Configuration (breaking) -- Reactor and Proactor terminology (breaking) -- Polly Resilience Pipeline v8 (breaking) -- Request Context enhancements -- InMemory options for testing -- Transport improvements (PostgreSQL, RabbitMQ, Kafka, AWS) -- Request ID is now an `Id` type - -**Migration Effort**: Most applications can be migrated in **1-4 hours**, depending on complexity. - -## Before You Start - -### Prerequisites - -1. **Backup your code**: Commit all changes to version control -2. **Review the release notes**: Read [Release Notes](https://github.com/BrighterCommand/Brighter/releases) for V10 -3. **Update test suite**: Ensure your tests are passing on V9 -4. **Check dependencies**: Review third-party package compatibility - -### Recommended Migration Approach - -1. **Upgrade in a feature branch**: Don't upgrade directly in main/master -2. **Address breaking changes first**: Fix compilation errors before adopting new features -3. **Test thoroughly**: Run your full test suite after each step -4. **Deploy to staging**: Validate in a non-production environment -5. **Monitor production**: Watch for issues after deployment - -## Step 1: Update Package References - -### Update NuGet Packages - -Update all Brighter packages to V10: - -```bash -# Core packages -dotnet add package Paramore.Brighter --version 10.0.0 -dotnet add package Paramore.Brighter.Extensions.DependencyInjection --version 10.0.0 - -# Transport packages (update as needed) -dotnet add package Paramore.Brighter.MessagingGateway.RMQ --version 10.0.0 -dotnet add package Paramore.Brighter.MessagingGateway.Kafka --version 10.0.0 -dotnet add package Paramore.Brighter.MessagingGateway.AWSSQS --version 10.0.0 - -# Outbox/Inbox packages (update as needed) -dotnet add package Paramore.Brighter.Outbox.MsSql --version 10.0.0 -dotnet add package Paramore.Brighter.Inbox.MsSql --version 10.0.0 -``` - -### Check for Package Conflicts - -```bash -# List all packages and check for conflicts -dotnet list package -``` - -## Step 2: Address Breaking Changes - -### 1. Nullable Reference Types - -**Breaking Change**: Nullable reference types are now enabled across all Brighter projects. - -**Migration Steps**: - -1. **Enable nullable reference types** in your project (if not already enabled): - -```xml - - - enable - - -``` - -2. **Address compiler warnings** in Commands, Events, and Handlers: - -**Before (V9)**: - -```csharp -public class CreatePersonCommand : Command -{ - public string Name { get; set; } // Warning: Non-nullable property - public string Email { get; set; } // Warning: Non-nullable property -} -``` - -**After (V10)**: - -```csharp -public class CreatePersonCommand : Command -{ - public required string Name { get; set; } // Required property - public required string Email { get; set; } // Required property - - // Or with constructor - public CreatePersonCommand(Guid id, string name, string email) : base(id) - { - Name = name; - Email = email; - } -} -``` - -3. **Update Message Mappers** to handle nullable warnings: - -```csharp -public class PersonCreatedMapper : IAmAMessageMapper -{ - public Message MapToMessage(PersonCreated request) - { - // Validate non-null properties - ArgumentNullException.ThrowIfNull(request.Name); - - var header = new MessageHeader( - messageId: request.Id, - topic: new RoutingKey("PersonCreated"), - messageType: MessageType.MT_EVENT - ); - - var body = new MessageBody(JsonSerializer.Serialize(request)); - return new Message(header, body); - } -} -``` - -**See also**: [Nullable Reference Types Documentation](/contents/NullableReferenceTypes.md) - -### 2. Simplified Configuration - -**Breaking Change**: Builder methods renamed for clarity. - -**Before (V9)**: -```csharp -services.AddBrighter() - .UseExternalBus(new RmqProducerRegistryFactory(...).Create()) - .AddServiceActivator(options => - { - options.Subscriptions = subscriptions; - options.ChannelFactory = new ChannelFactory(...); - }); -``` - -**After (V10)**: -```csharp -services.AddBrighter() - .AddProducers(options => - { - options.ProducerRegistry = new RmqProducerRegistryFactory(...).Create(); - }) - .AddConsumers(options => - { - options.Subscriptions = subscriptions; - options.ChannelFactory = new ChannelFactory(...); - }); -``` - -**Migration Steps**: - -1. Replace `UseExternalBus` with `AddProducers` -2. Replace `AddServiceActivator` with `AddConsumers` -3. Update property names: `ProducerRegistry` instead of passing directly - -### 3. Reactor and Proactor Terminology - -**Breaking Change**: The `runAsync` flag on Subscription has been renamed to `MessagePumpType`. - -**Before (V9)**: - -```csharp -var subscription = new Subscription( - new SubscriptionName("my-subscription"), - new ChannelName("my-channel"), - new RoutingKey("my.routing.key"), - runAsync: true // Old parameter -); -``` - -**After (V10)**: - -```csharp -var subscription = new Subscription( - new SubscriptionName("my-subscription"), - new ChannelName("my-channel"), - new RoutingKey("my.routing.key"), - messagePumpType: MessagePumpType.Proactor // New parameter -); -``` - -**Migration Table**: - -| V9 Parameter | V10 Parameter | Description | -|--------------|---------------|-------------| -| `runAsync: false` | `messagePumpType: MessagePumpType.Reactor` | Synchronous, blocking I/O | -| `runAsync: true` | `messagePumpType: MessagePumpType.Proactor` | Asynchronous, non-blocking I/O | - -**See also**: [Reactor and Proactor Documentation](/contents/ReactorAndProactor.md) - -### 4. Polly Resilience Pipeline - -**Breaking Change**: `TimeoutPolicyAttribute` is obsolete. Use `UseResiliencePipeline` attribute. - -**Before (V9)**: - -```csharp -public class MyHandler : RequestHandlerAsync -{ - [TimeoutPolicy(milliseconds: 5000, step: 1)] - public override async Task HandleAsync( - MyCommand command, - CancellationToken cancellationToken = default) - { - // Handler logic - return await base.HandleAsync(command, cancellationToken); - } -} -``` - -**After (V10)**: - -1. **Define a Resilience Pipeline**: - -```csharp -var resiliencePipelineRegistry = new ResiliencePipelineRegistry(); -resiliencePipelineRegistry.TryAddBuilder>( - "MyPipeline", - (builder, context) => - { - builder.AddTimeout(TimeSpan.FromSeconds(5)); - builder.AddRetry(new RetryStrategyOptions - { - MaxRetryAttempts = 3, - Delay = TimeSpan.FromMilliseconds(100) - }); - }); -``` - -2. **Use the new attribute**: - -```csharp -public class MyHandler : RequestHandlerAsync -{ - [UseResiliencePipeline(policy: "MyPipeline", step: 1)] - public override async Task HandleAsync( - MyCommand command, - CancellationToken cancellationToken = default) - { - // Handler logic - return await base.HandleAsync(command, cancellationToken); - } -} -``` - -3. **Register the pipeline** with Brighter: - -```csharp -services.AddBrighter(options => -{ - options.PolicyRegistry = resiliencePipelineRegistry; -}); -``` - -**See also**: [Policy Retry and Circuit Breaker Documentation](/contents/PolicyRetryAndCircuitBreaker.md) - -### 5. Request Context Interface Changes - -**Breaking Change**: `IRequestContext` interface has new properties. - -**New Properties**: - -- `PartitionKey`: Set message partition keys dynamically -- `CustomHeaders`: Add custom headers via request context -- `ResilienceContext`: Integration with Polly resilience pipeline -- `OriginatingMessage`: Access the original message (for consumers) - -**Migration**: Most code should not be affected unless you implement `IRequestContext` directly. - -**If you implement `IRequestContext`** (rare): - -```csharp -public class MyRequestContext : IRequestContext -{ - public Guid Id { get; set; } - public ISpan Span { get; set; } - public Dictionary Bag { get; set; } - - // V10: Add new properties - public string? PartitionKey { get; set; } - public Dictionary CustomHeaders { get; set; } = new(); - public ResilienceContext? ResilienceContext { get; set; } - public Message? OriginatingMessage { get; set; } -} -``` - -**Using new properties**: - -```csharp -public class MyHandler : RequestHandlerAsync -{ - public override async Task HandleAsync( - MyCommand command, - CancellationToken cancellationToken = default) - { - // Set partition key for message routing - Context.PartitionKey = command.TenantId; - - // Add custom headers - Context.CustomHeaders["X-Correlation-Id"] = command.CorrelationId; - - // Access originating message (for consumers) - if (Context.OriginatingMessage != null) - { - var receivedTimestamp = Context.OriginatingMessage.Header.TimeStamp; - } - - return await base.HandleAsync(command, cancellationToken); - } -} -``` - -### 6. Request Id and CorrelationId type change - -In V9 the `Request` type used a `Guid` to represent the identity of a `Command` or `Event`. In V10, as part of a move away from primitives, we have changed this to be a type `Id`. An `Id` can be constructed from a `string` using its `ToString()` method. If you have used a `Guid` then you will need to turn the `Guid` into a `string`. - -Within `IRequest` both `Id` and `CorrelationId` both use the type `Id` in V10. - -If you were using a `Guid` to create a random identity, you can just use `Id.Random()` instead which has the same behavior. - -Before: - -```csharp - -class MyCommand() : Command(Guid.NewGuid()) -{ - public string Value { get; set; } = string.Empty; -} - -``` - -After: - -```csharp - -class MyCommand() : Command(Id.Random()) -{ - public string Value { get; set; } = string.Empty; -} - -``` - -## Step 3: Adopt New Features (Optional) - -### 1. Cloud Events Support - -V10 adds full Cloud Events specification support. - -**Benefits**: - -- Standardized event metadata -- Better interoperability with other systems -- Rich event context information - -**Adoption Steps**: - -1. **Update Publication** to include Cloud Events properties: - -```csharp -new RmqPublication -{ - Topic = new RoutingKey("PersonCreated"), - CloudEventsType = new CloudEventsType("io.paramore.person.created"), - Source = new Uri("https://api.example.com/persons"), - Subject = "person/created", - MakeChannels = OnMissingChannel.Create -} -``` - -2. **Use Cloud Events headers** in your mapper (optional): - -```csharp -public class PersonCreatedMapper : IAmAMessageMapper -{ - public Message MapToMessage(PersonCreated request, Publication publication) - { - var header = new MessageHeader( - messageId: request.Id, - topic: publication.Topic, - messageType: MessageType.MT_EVENT, - type: publication.CloudEventsType, // Use Cloud Events type - source: publication.Source, - subject: publication.Subject - ); - - var body = new MessageBody(JsonSerializer.Serialize(request)); - return new Message(header, body); - } -} -``` - -**See also**: [Cloud Events Documentation](/contents/CloudEventsSupport.md) - -### 2. Default Message Mappers - -V10 allows you to omit message mappers for simple JSON serialization. - -**Benefits**: - -- Less boilerplate code -- Faster development -- Still supports custom mappers for complex scenarios - -**Before (V9) - Required Mapper**: - -```csharp -public class PersonCreatedMapper : IAmAMessageMapper -{ - public Message MapToMessage(PersonCreated request) - { - var header = new MessageHeader( - messageId: request.Id, - topic: new RoutingKey("PersonCreated"), - messageType: MessageType.MT_EVENT - ); - - var body = new MessageBody(JsonSerializer.Serialize(request)); - return new Message(header, body); - } - - public PersonCreated MapToRequest(Message message) - { - return JsonSerializer.Deserialize(message.Body.Value) - ?? throw new InvalidOperationException("Failed to deserialize"); - } -} - -// Register mapper -messageMapperRegistry.Register(); -``` - -**After (V10) - No Mapper Needed**: - -```csharp -// No mapper registration needed for simple JSON serialization! -// Brighter uses JsonMessageMapper by default - -// Just define your message -public class PersonCreated : Event -{ - public string Name { get; set; } - public string Email { get; set; } -} - -// Publish directly -await commandProcessor.PublishAsync(new PersonCreated -{ - Name = "Alice", - Email = "alice@example.com" -}); -``` - -**When to use custom mappers**: - -- Complex transformations -- Non-JSON formats (XML, Protobuf, etc.) -- Transform pipelines (encryption, compression, claim check) -- Custom header mapping - -**See also**: [Message Mappers Documentation](/contents/MessageMappers.md) - -### 3. Dynamic Message Deserialization - -V10 supports multiple message types on the same channel. - -**Benefits**: - -- Content-based routing -- Flexible message processing -- Better use of infrastructure - -**Example**: - -```csharp -new KafkaSubscription( - new SubscriptionName("task-state-subscription"), - channelName: new ChannelName("task.state"), - routingKey: new RoutingKey("task.update"), - getRequestType: message => message switch - { - var m when m.Header.Type == new CloudEventsType("io.paramore.task.created") - => typeof(TaskCreated), - var m when m.Header.Type == new CloudEventsType("io.paramore.task.updated") - => typeof(TaskUpdated), - var m when m.Header.Type == new CloudEventsType("io.paramore.task.deleted") - => typeof(TaskDeleted), - _ => throw new ArgumentException( - $"No type mapping found for message with type {message.Header.Type}", - nameof(message)) - }, - groupId: "task-consumer-group", - messagePumpType: MessagePumpType.Proactor -); -``` - -**See also**: [Dynamic Deserialization Documentation](/contents/DynamicDeserialization.md) - -### 4. OpenTelemetry Semantic Conventions - -V10 adopts OpenTelemetry Semantic Conventions for messaging. - -**Impact**: Trace spans will have different names and attributes than V9. - -**Benefits**: - -- Standard messaging conventions -- Better integration with APM tools -- Consistent telemetry across systems - -**Migration**: - -1. **Update dashboards and alerts** to use new span names -2. **Review trace queries** for compatibility -3. **Test observability** in staging environment - -**V9 Span Names**: - -- `Paramore.Brighter.CommandProcessor.Send` -- `Paramore.Brighter.MessagePump.Receive` - -**V10 Span Names** (OTel Semantic Conventions): - -- `messaging.send` -- `messaging.receive` -- `messaging.process` - -**See also**: [Telemetry Documentation](/contents/Telemetry.md) - -### 5. InMemory Options for Testing - -V10 provides comprehensive InMemory implementations for testing. - -**Benefits**: - -- Fast test execution -- No external dependencies -- Easy CI/CD integration - -**Example**: - -```csharp -// Test setup with InMemory components -var internalBus = new InternalBus(); - -services.AddBrighter(options => -{ - options.HandlerLifetime = ServiceLifetime.Scoped; -}) -.AddProducers(options => -{ - var publication = new Publication { Topic = new RoutingKey("TestTopic") }; - options.ProducerRegistry = new InMemoryProducerRegistryFactory( - internalBus, - new[] { publication }, - InstrumentationOptions.All - ).Create(); - options.Outbox = new InMemoryOutbox(TimeProvider.System); -}) -.AddConsumers(options => -{ - options.Inbox = new InboxConfiguration( - new InMemoryInbox(TimeProvider.System), - InboxConfiguration.NoActionOnExists - ); - options.Subscriptions = subscriptions; - options.ChannelFactory = new InMemoryChannelFactory(internalBus, TimeProvider.System); -}) -.UseScheduler(new InMemorySchedulerFactory()) -.AutoFromAssemblies(); -``` - -**See also**: [InMemory Options Documentation](/contents/InMemoryOptions.md) - -## Step 4: Test Your Migration - -### Unit Tests - -1. **Run existing unit tests**: - -```bash -dotnet test -``` - -2. **Address test failures**: - - Update mocks for `IRequestContext` new properties - - Update assertions for OpenTelemetry span names - - Fix nullable reference warnings - -### Integration Tests - -1. **Test with InMemory components** (fast): - -```csharp -[Fact] -public async Task Should_Process_Message_With_V10_Components() -{ - // Arrange - var internalBus = new InternalBus(); - var serviceProvider = BuildServiceProvider(internalBus); - var commandProcessor = serviceProvider.GetRequiredService(); - - // Act - await commandProcessor.PublishAsync(new PersonCreated - { - Name = "Alice", - Email = "alice@example.com" - }); - - // Assert - var messages = internalBus.Stream(new RoutingKey("PersonCreated")); - Assert.NotEmpty(messages); -} -``` - -2. **Test with real transports** (slower, more complete): - -```bash -# Start dependencies (Docker Compose) -docker-compose up -d rabbitmq postgres - -# Run integration tests -dotnet test --filter Category=Integration -``` - -### Performance Testing - -Compare V9 and V10 performance: - -```csharp -[Benchmark] -public async Task V10_MessageProcessing() -{ - for (int i = 0; i < 1000; i++) - { - await _commandProcessor.SendAsync(new TestCommand { Value = i }); - } -} -``` - -Expected: V10 should have similar or better performance due to optimizations. - -## Step 5: Deploy to Staging - -### Pre-Deployment Checklist - -- [ ] All tests passing -- [ ] Code review completed -- [ ] Documentation updated -- [ ] Release notes reviewed -- [ ] Rollback plan prepared - -### Deployment Steps - -1. **Deploy to staging environment** -2. **Run smoke tests** -3. **Monitor metrics**: - - Message processing latency - - Error rates - - Resource usage (CPU, memory) -4. **Check logs** for warnings or errors -5. **Validate telemetry** (traces, metrics) - -### Monitoring - -Watch for: -- Increased error rates -- Performance degradation -- OpenTelemetry trace issues -- Null reference exceptions - -## Step 6: Deploy to Production - -### Production Deployment - -1. **Deploy during low-traffic period** -2. **Use blue-green or canary deployment** if possible -3. **Monitor closely** for first 24 hours -4. **Have rollback plan ready** - -### Post-Deployment - -- Review logs for issues -- Check application metrics -- Validate message processing -- Monitor OpenTelemetry traces - -## Common Migration Issues - -### Issue 1: Nullable Reference Warnings - -**Symptom**: CS8618, CS8600, CS8602 warnings - -**Solution**: See [Nullable Reference Types Documentation](/contents/NullableReferenceTypes.md) - -### Issue 2: Method Not Found - -**Symptom**: `UseExternalBus` method not found - -**Solution**: Replace with `AddProducers` - -### Issue 3: Property Not Found on Subscription - -**Symptom**: `runAsync` property does not exist - -**Solution**: Use `messagePumpType: MessagePumpType.Proactor` or `MessagePumpType.Reactor` - -### Issue 4: TimeoutPolicy Not Working - -**Symptom**: `TimeoutPolicyAttribute` marked as obsolete - -**Solution**: Migrate to `UseResiliencePipeline` with Polly v8 - -### Issue 5: Telemetry Spans Changed - -**Symptom**: Dashboard queries not finding spans - -**Solution**: Update queries to use OpenTelemetry Semantic Convention names - -## Rollback Plan - -If you need to roll back to V9: - -1. **Revert package versions**: - -```bash -dotnet add package Paramore.Brighter --version 9.x.x -``` - -2. **Revert code changes**: - -```bash -git revert -``` - -3. **Redeploy V9 version** - -4. **Investigate issues** before attempting V10 migration again - -## Getting Help - -### Resources - -- [Brighter Documentation](https://brightercommand.github.io/Brighter/) -- [GitHub Issues](https://github.com/BrighterCommand/Brighter/issues) -- [Release Notes](https://github.com/BrighterCommand/Brighter/releases) -- [Stack Overflow](https://stackoverflow.com/questions/tagged/brighter) - -### Reporting Issues - -If you encounter issues during migration: - -1. **Check existing issues**: Search [GitHub Issues](https://github.com/BrighterCommand/Brighter/issues) -2. **Create a new issue** with: - - V9 and V10 versions - - Minimal reproduction example - - Error messages and stack traces - - Environment details (.NET version, OS, transport) - -## Summary - -Migrating to Brighter V10 involves: - -1. **Update packages** to V10 -2. **Address breaking changes**: - - Nullable reference types - - Simplified configuration API - - Reactor/Proactor terminology - - Polly Resilience Pipeline -3. **Adopt new features** (optional): - - Cloud Events - - Default message mappers - - Dynamic deserialization - - OpenTelemetry conventions - - InMemory testing options -4. **Test thoroughly** -5. **Deploy to staging**, then production -6. **Monitor** and address any issues - -Most migrations can be completed in **1-4 hours**. The breaking changes are straightforward, and V10 provides significant improvements in features, performance, and developer experience. - -Good luck with your migration! 🚀 +__V10__ \ No newline at end of file From c54d12a58942f7f71f9a3e04e276e5131a174f04 Mon Sep 17 00:00:00 2001 From: Rafael Lillo <7280959+lillo42@users.noreply.github.com> Date: Thu, 23 Jul 2026 20:51:21 +0100 Subject: [PATCH 2/3] KafkaConfiguration: document KIP-848 consumer group protocol (BrighterCommand/Brighter#4233) - Add Consumer Group Protocol (KIP-848) section covering IGroupProtocol, ClassicGroupProtocol, ConsumerGroupProtocol, back-fill behavior, the no-sharing contract, static membership and Validate caveats - Mark SessionTimeout and PartitionAssignmentStrategy subscription options as deprecated in favor of GroupProtocol - Update Partition Assignment Strategy examples to use ClassicGroupProtocol --- contents/KafkaConfiguration.md | 551 ++++++++++++++++++++++++++++++++- 1 file changed, 550 insertions(+), 1 deletion(-) diff --git a/contents/KafkaConfiguration.md b/contents/KafkaConfiguration.md index 8694137..858faf1 100644 --- a/contents/KafkaConfiguration.md +++ b/contents/KafkaConfiguration.md @@ -1 +1,550 @@ -__KAFKA__ \ No newline at end of file +# Kafka Configuration + +## General + +Kafka is OSS message-oriented-middleware and is [well documented](https://kafka.apache.org/documentation/#gettingStarted). Brighter handles the details of sending to or receiving from Kafka. You may find it useful to understand the [building blocks](https://kafka.apache.org/documentation/#introduction) of the protocol. Brighter's Kafka support is implemented on top of the Confluent .NET client, and you might find the [documentation for the .NET client](https://docs.confluent.io/kafka-clients/dotnet/current/overview.html) helpful when debugging, but you should not have to interact with it directly to use Brighter (although we expose many of its configuration options). + +Kafka has two main roles: + +- **Producer**: A producer sends events to a **Topic** on a Kafka broker. +- **Consumer**: A consumer reads events from a **Topic** on a Kafka broker. + +**Topics** are append-only streams of events. Multiple producers can write to a topic, and multiple consumers can read from one. A **consumer** uses an **offset** into the stream to indicate the event it wants to read. Kafka does not delete an event from the stream when it is ack'd by the consumer; instead a **consumer** increments its **offset** once an item has been read so that it can avoid processing the same event twice. See [Offset Management](#offset-management) for more on how Brighter manages **consumer offsets**. As a result the lifetime of events on a stream is instead a configuration setting for the stream. + +As a **consumer** manages an **offset** to record events that is has read, you cannot scale an application that wishes to consume a **topic** by increasing the number of **consumers**--they don't share an offset--without partitioning the **topic**. If you supply a **partition key**, a **partition** uses consistent hashing to slice a **topic** into a number of streams; otherwise it will use round-robin. See [this documentation](https://jaceklaskowski.gitbooks.io/apache-kafka/content/kafka-producer-internals-DefaultPartitioner.html) for more. Each **partition** is only read by a single **consumer** within the application. All of the consumers for an application should share the same group id, called a **consumer group** in Kafka. As each **consumer** tracks the **offset** for the **partitions** it is reading, it is possible to have multiple **consumers** read and process the same **topic**. + +A **consumer** may read from *multiple* **partitions**, but only one **consumer** may read from a **partition** at one time in a given **consumer group**. Kafka will assign partitions across the pool of consumers for the **consumer group**. When the pool changes, a **rebalance** occurs, which may mean that a consumer changes the **partition** that it is assigned within the **consumer group**. Brighter favors *sticky assignment of partitions* to avoid unnecessary churn of partitions. + +In addition to the Producer API and Consumer API Kafka streams have features such as the Streams API and the Connect API. We do not use either of these from Brighter. + +## Connection + +The Connection to Kafka is provided by an **KafkaMessagingGatewayConnection** which allows you to configure the following: + +- **BootstrapServers**: A **bootstrap** server is a well-known broker through which we discover the servers in the Kafka cluster that we can connect to. You should supply a comma-separated list of host and port pairs. These are the addresses of the Kafka brokers in the "bootstrap" Kafka cluster. +- **Debug**: A comma-separated list of debug contexts to enable. Producer: broker, topic, msg. Consumer: consumer, cgrp, topic, fetch. +- **Name**: An identifier to use for the client. +- **SaslMechanisms**: If any, what is the protocol used for authenticated connection to the Kafka broker: plain, scram-sha-256, scram-sha-256, gssapi (kerberos), oauthbearer +- **SaslKerberosName**: If using kerberos, what is the connection name. +- **SaslUsername**: SASL username for use with PLAIN and SASL-SCRAM +- **SaslPassword**: SASL password for use with PLAIN and SASL-SCRAM +- **SecurityProtocol**: How are messages between client and server encrypted, if at all: plaintext, ssl, saslplaintext, saslssl +- **SslCaLocation**: Where is the CA certificate located (see [here](https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/csharp.html) for guidance). +- **SslKeystoreLocation**: Path to the client's keystore +- **SslKeystorePassword**: Password for the client's keystore + +The following code connects to a local Kafka instance (for development): + +``` csharp + services.AddBrighter(...) + .AddProducers((configure) => + { + configure.ProducerRegistry = new KafkaProducerRegistryFactory( + new KafkaMessagingGatewayConfiguration() + { + Name = "paramore.brighter.greetingsender", + BootStrapServers = new[] {"localhost:9092"} + }, + ...//publication, see below + ) + .Create(); + }) + ... + +``` + +The following code connects to a remote Kafka instance. The settings here will depend on how your production broker is configured for access. We show getting secrets from environment variables for simplicity, again you will need to adjust this for your approach to secrets management: + +``` csharp + services.AddBrighter(...) + .AddProducers((configure) => + { + configure.ProducerRegistry = new KafkaProducerRegistryFactory( + new KafkaMessagingGatewayConfiguration() + { + Name = "paramore.brighter.greetingsender", + BootStrapServers = new[] { Environment.GetEnvironmentVariable("BOOSTRAP_SERVER")}, + SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol.SaslSsl, + SaslMechanisms = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism.Plain, + SaslUsername = Environment.GetEnvironmentVariable("SASL_USERNAME"), + SaslPassword = Environment.GetEnvironmentVariable("SASL_PASSWORD"), + SslCaLocation = RuntimeInformation.IsOSPlatform(OSPlatform.OSX) ? "/usr/local/etc/openssl@1.1/cert.pem" : null; + }, + ...//publication, see below + ) + .Create(); + }) + ... + +``` + +## Publication + +For more on a *Publication* see the material on an *Add Producers* in [Basic Configuration](/contents/BrighterBasicConfiguration.md#using-an-external-bus). + +We allow you to configure properties for both Brighter and the Confluent .NET client. Because there are many properties on the Confluent .NET Client we also configure a callback to let you inspect and modify the configuration that we will pass to the client if you so desire. This can be used to add properties we do not support or adjust how we set them. + +- **Replication**: how many ISR nodes must receive the record before the producer can consider the write successful. Default is Acks.All. +- **BatchNumberMessages**: Maximum number of messages batched in one MessageSet. Default is 10. +- **EnableIdempotence**: Messages are produced once only. Will adjust the following if not set: `max.in.flight.requests.per.connection=5` (must be less than or equal to 5), `retries=INT32_MAX` (must be greater than 0), `acks=all`, `queuing.strategy=fifo`. Default is true. +- **LingerMs**: Maximum time, in milliseconds, for buffering data on the producer queue. Default is 5. +- **MessageSendMaxRetries**: How many times to retry sending a failing MessageSet. Note: retrying may cause reordering, set the max in flight to 1 if you need ordering by when sent. Default is 3. +- **MessageTimeoutMs**: Local message timeout. This value is only enforced locally and limits the time a produced message waits for successful delivery. A time of 0 is infinite. Default is 5000. +- **MaxInFlightRequestsPerConnection**: Maximum number of in-flight requests the client will send. We default this to 1, so as to allow retries to not de-order the stream. +- **NumPartitions**: How many partitions for this topic. We default to 1. +- **Partitioner**: How do we partition? Defaults to Partitioner.ConsistentRandom. +- **QueueBufferingMaxMessages**: Maximum number of messages allowed on the producer queue. Defaults to 10. +- **QueueBufferingMaxKbytes**: Maximum total message size sum allowed on the producer queue. Defaults to 1048576 bytes (so for 10 messages about 104Kb per message). +- **ReplicationFactor**: What is the replication factor? How many nodes is the topic copied to on the broker? Defaults to 1. +- **RetryBackoff**: The backoff time before retrying a message send. Defaults to 100. +- **RequestTimeoutMs**: The ack timeout of the producer request. This value is only enforced by the broker and relies on Replication being != AcksEnum.None. Defaults to 500. +- **TopicFindTimeoutMs**: How long to wait when asking for topic metadata. Defaults to 5000. +- **TransactionalId**: The unique identifier for this producer, used with transactions + +The following example shows how a *Publication* might be configured: + +``` csharp + services.AddBrighter(...) + .AddProducers((configure) => + { + configure.ProducerRegistry = new KafkaProducerRegistryFactory( + ...,//connection see above + new KafkaPublication[] {new KafkaPublication() + { + Topic = new RoutingKey("MyTopicName"), + NumPartitions = 3, + ReplicationFactor = 3, + MessageTimeoutMs = 1000, + RequestTimeoutMs = 1000, + MakeChannels = OnMissingChannel.Create + }} + ).Create(); + }) + +``` + +### Configuration Callback + +The Confluent .NET client has a range of configuration options. Some of those can be controlled through the publication. But, to allow you the full range of configuration options for the Confluent client, including new options that may appear, we provide a callback on the **KafkaProducerRegistryFactory**. The registry exposes a method, **SetConfigHook(Action hook)**. The method takes a *delegate* (you can pass a lambda). Your delegate will be called with the *proposed* ProducerConfig (taking into account the *Publication* settings). You can adjust additional parameters at this point. + +You can use it as follows: + +``` csharp + + var publication = new KafkaPublication() + { + Topic = new RoutingKey("MyTopicName"), + NumPartitions = 3, + ReplicationFactor = 3, + MessageTimeoutMs = 1000, + RequestTimeoutMs = 1000, + MakeChannels = OnMissingChannel.Create + }; + publication.SetConfigHook(config => config.EnableGaplessGuarantee = true) + + services.AddBrighter(...) + .AddProducers((configure) => + { + configure.ProducerRegistry = new KafkaProducerRegistryFactory( + ...,//connection see above + new KafkaPublication() {publication} + ).Create(); + }) + +``` + +### Kafka Topic Auto Create + +Brighter uses the Kafka AdminClient for topic creation. For this to work as expected you should set the server property of **auto.create.topics.enable** to **false**; otherwise the topic will be auto-created with the values defined by your server for new topics, such as the number of partitions. This error can be insidious because your code will still work against this topic, but without inspection you will not observe that its properties do not match those requested. + +If you want to specify the topic through Brighter, or through your own IaaS code, we recommend always setting this setting to false; we recommend only setting it to true if you tell Brighter to assume that the infrastructure exists, as it will then be created on the first write. + +## Subscription + +For more on a *Subscription* see the material on configuring *Service Activator* in [Basic Configuration](/contents/BrighterBasicConfiguration.md#configuring-the-service-activator). + +We support a number of Kafka specific *Subscription* options: + +- **CommitBatchSize**: We commit processed work (marked as acked or rejected) when a batch size worth of work has been completed (see [below](#offset-management)). Defaults to 10. +- **ConfigHook**: Allows you to modify the Kafka client configuration before a consumer is created. Used to set properties that Brighter does not expose. See [Configuration Callback](#configuration-callback) below. +- **GroupId**: Only one consumer in a group can read from a partition at any one time; this preserves ordering. We do not default this value, and expect you to set it. +- **GroupProtocol**: Selects the Kafka consumer group protocol: `ClassicGroupProtocol` (default) or `ConsumerGroupProtocol` (KIP-848). See [Consumer Group Protocol (KIP-848)](#consumer-group-protocol-kip-848) below. Defaults to null, which uses the classic protocol. +- **IsolationLevel**: Default to read only committed messages, change if you want to read uncommitted messages. May cause duplicates. Defaults to ReadCommitted. +- **MaxPollInterval**: How often the consumer needs to poll for new messages to be considered alive, polling greater than this interval triggers a re-balance. Defaults to 300000ms (5 minutes). +- **NumPartitions**: How many partitions does the topic have? Used for topic creation, if required. Defaults to 1. +- **OffsetDefault**: What do we do if there is no offset stored for this consumer. Defaults to AutoOffsetReset.Earliest - Begin reading the stream from the start. Options include AutoOffsetReset.Latest - Start from now i.e. only consume messages after we start and AutoOffsetReset.Error - which considers it an error if no reset is found +- **PartitionAssignmentStrategy**: How do partitions get assigned to consumers in the group? Defaults to RoundRobin for even distribution. See [Partition Assignment Strategy](#partition-assignment-strategy) below. **Deprecated**: set `PartitionAssignmentStrategy` on a `ClassicGroupProtocol` via **GroupProtocol** instead. +- **ReadCommittedOffsetsTimeOut**: How long before attempting to read back committed offsets (mainly used in debugging) is an error. Defaults to 5000ms. +- **ReplicationFactor**: What is the replication factor? How many nodes is the topic copied to on the broker? Defaults to 1. Used for topic creation if required. +- **SessionTimeout**: If Kafka does not receive a heartbeat from the consumer within this time window, trigger a re-balance. Defaults to 10000ms (10 seconds). **Deprecated**: set `SessionTimeout` on a `ClassicGroupProtocol` via **GroupProtocol** instead. +- **SweepUncommittedOffsetsInterval**: The interval at which we sweep, looking for offsets that have not been flushed (see [below](#offset-management)). Defaults to 30000ms (30 seconds). + +The following example shows how a subscription might be configured: + +``` csharp + private static void ConfigureBrighter(HostBuilderContext hostContext, IServiceCollection services) +{ + var subscriptions = new KafkaSubscription[] + { + new KafkaSubscription( + subscriptionName: new SubscriptionName("paramore.example.greeting"), + channelName: new ChannelName("greeting.event"), + routingKey: new RoutingKey("greeting.event"), + groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID"), + timeOut: TimeSpan.FromMilliseconds(100), + commitBatchSize: 5, + sweepUncommittedOffsetsInterval: TimeSpan.FromMilliseconds(3000), + messagePumpType: MessagePumpType.Reactor + ) + }; + + //create the gateway + var consumerFactory = new KafkaMessageConsumerFactory( + new KafkaMessagingGatewayConfiguration {...} // see connection information above + ); + + services.AddConsumers(options => + { + options.Subscriptions = subscriptions; + options.ChannelFactory = new ChannelFactory(consumerFactory); + }).AutoFromAssemblies(); + + + services.AddHostedService(); +} +``` + +### Configuration Callback + +Similar to producers, the Confluent .NET client for consumers has a range of configuration options. Some of those can be controlled through the subscription. But, to allow you the full range of configuration options for the Confluent client, including new options that may appear, we provide a **configHook** parameter on **KafkaSubscription**. + +The **configHook** takes a *delegate* (you can pass a lambda). Your delegate will be called with the *proposed* ConsumerConfig (taking into account the *Subscription* settings). You can adjust additional parameters at this point. + +You can use it as follows: + +``` csharp +var subscription = new KafkaSubscription( + subscriptionName: new SubscriptionName("paramore.example.greeting"), + channelName: new ChannelName("greeting.event"), + routingKey: new RoutingKey("greeting.event"), + groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID"), + timeOut: TimeSpan.FromMilliseconds(100), + commitBatchSize: 5, + sweepUncommittedOffsetsInterval: TimeSpan.FromMilliseconds(3000), + messagePumpType: MessagePumpType.Reactor, + configHook: config => + { + // Customize the Confluent Consumer configuration + config.FetchMinBytes = 1024; + config.FetchWaitMaxMs = 500; + config.StatisticsIntervalMs = 60000; // Enable statistics + } +); + +var subscriptions = new KafkaSubscription[] { subscription }; + +//create the gateway +var consumerFactory = new KafkaMessageConsumerFactory( + new KafkaMessagingGatewayConfiguration {...} // see connection information above +); + +services.AddConsumers(options => +{ + options.Subscriptions = subscriptions; + options.ChannelFactory = new ChannelFactory(consumerFactory); +}).AutoFromAssemblies(); +``` + +### Common Configuration Callback Use Cases + +The **configHook** is useful for: + +**1. Fine-tuning fetch behavior**: + +``` csharp +configHook: config => +{ + config.FetchMinBytes = 1024; // Minimum data per fetch request + config.FetchWaitMaxMs = 100; // Max time to wait for fetch.min.bytes + config.FetchMaxBytes = 52428800; // Maximum data for all partitions +} +``` + +**2. Enabling consumer statistics**: + +``` csharp +configHook: config => +{ + config.StatisticsIntervalMs = 30000; // Emit stats every 30 seconds +} +``` + +**3. Customizing security settings**: + +``` csharp +configHook: config => +{ + config.EnableSslCertificateVerification = true; + config.SslEndpointIdentificationAlgorithm = SslEndpointIdentificationAlgorithm.Https; +} +``` + +**4. Adjusting partition assignment**: + +``` csharp +configHook: config => +{ + // Note: PartitionAssignmentStrategy should typically be set via the subscription parameter + // But you can override it here if needed + config.SessionTimeoutMs = 45000; // Increase for slow rebalances +} +``` + +**5. Debugging and monitoring**: + +``` csharp +configHook: config => +{ + config.Debug = "consumer,cgrp,topic,fetch"; // Enable debug logging + config.StatisticsIntervalMs = 10000; // Frequent statistics for debugging +} +``` + +### Partition Assignment Strategy + +Kafka distributes partitions across consumers in a consumer group using a partition assignment strategy. Brighter provides control over this strategy through the **partitionAssignmentStrategy** parameter on the subscription, or the **PartitionAssignmentStrategy** property on a `ClassicGroupProtocol` (see [Consumer Group Protocol (KIP-848)](#consumer-group-protocol-kip-848)). + +> **Deprecated**: The **partitionAssignmentStrategy** parameter and the **PartitionAssignmentStrategy** property on `KafkaSubscription` are obsolete and will be removed in a future version. Configure the strategy on a `ClassicGroupProtocol` assigned to the subscription's **GroupProtocol** property instead. Note that partition assignment strategies only apply to the classic protocol; with KIP-848 partition assignment is broker-driven. + +**Available Strategies**: + +- **RoundRobin** (default): Distributes partitions evenly across consumers in a round-robin fashion. This is the most balanced distribution but may cause all partitions to be reassigned during rebalances. +- **Range**: Assigns contiguous partition ranges to consumers. Useful for co-locating related partitions on the same consumer. +- **CooperativeSticky**: Not supported with Brighter's manual offset commits (will throw an exception). + +**Why RoundRobin is the Default**: + +Brighter uses **RoundRobin** as the default strategy because: +1. It provides the most even distribution of partitions across consumers +2. It works reliably with Brighter's manual offset commit strategy +3. It's straightforward to reason about for most use cases + +**Using Range Strategy**: + +The **Range** strategy is useful when you want to co-locate partitions on the same consumer: + +``` csharp +var subscription = new KafkaSubscription( + subscriptionName: new SubscriptionName("paramore.example.greeting"), + channelName: new ChannelName("greeting.event"), + routingKey: new RoutingKey("greeting.event"), + groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID") +) +{ + GroupProtocol = new ClassicGroupProtocol + { + PartitionAssignmentStrategy = PartitionAssignmentStrategy.Range + } +}; +``` + +**CooperativeSticky Limitation**: + +The **CooperativeSticky** strategy is not supported when using manual offset commits (which Brighter requires for its at-least-once delivery guarantees). Attempting to use it will throw an `ArgumentOutOfRangeException`: + +``` csharp +// This will throw an exception: +var subscription = new KafkaSubscription( + // ... +) +{ + GroupProtocol = new ClassicGroupProtocol + { + PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky // ❌ Not supported + } +}; +``` + +This is due to a [known issue in librdkafka](https://github.com/confluentinc/librdkafka/issues/4059) where CooperativeSticky doesn't work correctly with manual offset management. + +### Consumer Group Protocol (KIP-848) + +[KIP-848](https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol) is Kafka's next-generation consumer group protocol: group membership and partition assignment become broker-driven, which significantly reduces rebalance times and removes the "stop-the-world" rebalances of the classic protocol. It is generally available from Apache Kafka 4.0 (which also requires KRaft mode, with no ZooKeeper). + +Brighter supports both protocols through the **GroupProtocol** property on `KafkaSubscription`, which takes an `IGroupProtocol` implementation: + +- **ClassicGroupProtocol** (default): Kafka's original consumer group protocol, where partition assignment is computed by the clients. Carries `SessionTimeout`, `HeartbeatInterval`, and `PartitionAssignmentStrategy`. +- **ConsumerGroupProtocol**: The KIP-848 consumer protocol (`group.protocol=consumer`), where partition assignment is computed by the broker. Carries `GroupRemoteAssignor` (the broker-side assignor name) and `GroupInstanceId` (static membership). + +**Using the classic protocol explicitly**: + +``` csharp +var subscription = new KafkaSubscription( + subscriptionName: new SubscriptionName("paramore.example.greeting"), + channelName: new ChannelName("greeting.event"), + routingKey: new RoutingKey("greeting.event"), + groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID") +) +{ + GroupProtocol = new ClassicGroupProtocol + { + SessionTimeout = TimeSpan.FromSeconds(30), + HeartbeatInterval = TimeSpan.FromSeconds(10), + PartitionAssignmentStrategy = PartitionAssignmentStrategy.RoundRobin + } +}; +``` + +When you don't supply a **GroupProtocol**, Brighter uses a `ClassicGroupProtocol` and back-fills any null properties from the consumer's constructor parameters — `SessionTimeout` (default 10 seconds) and `PartitionAssignmentStrategy` (default `RoundRobin`) — preserving the existing behavior. If you do supply a `ClassicGroupProtocol`, any values you have set are preserved; only null properties are back-filled. + +> **Warning**: The back-fill mutates the `ClassicGroupProtocol` instance in place. Do not share one instance across subscriptions — with a shared instance, the first consumer's back-filled values would silently apply to the second subscription as well. The instance is also applied to every consumer created from the subscription (one per performer). + +**Using the KIP-848 consumer protocol**: + +``` csharp +var subscription = new KafkaSubscription( + subscriptionName: new SubscriptionName("paramore.example.greeting"), + channelName: new ChannelName("greeting.event"), + routingKey: new RoutingKey("greeting.event"), + groupId: Environment.GetEnvironmentVariable("KAFKA_GROUPID") +) +{ + GroupProtocol = new ConsumerGroupProtocol() +}; +``` + +`ConsumerGroupProtocol` deliberately does **not** set `session.timeout.ms`, `heartbeat.interval.ms`, or `partition.assignment.strategy`: librdkafka rejects those properties when `group.protocol=consumer`. Heartbeats, session management, and partition assignment are broker-driven under KIP-848. + +**Broker requirement**: The KIP-848 consumer protocol requires a broker that supports it (Apache Kafka 4.0 or later). Using `ConsumerGroupProtocol` against an older broker will fail. + +**Static membership**: + +`ConsumerGroupProtocol.GroupInstanceId` enables Kafka static membership (`group.instance.id`). Static membership requires a **unique** id per consumer instance. Because the subscription's `GroupProtocol` instance is applied to every consumer created from that subscription (one per performer), all of those consumers would register the same static id and collide on group join. Only set `GroupInstanceId` when the subscription has a single performer; to use static membership with multiple performers, assign a unique `group.instance.id` per consumer via the **configHook** instead: + +``` csharp +configHook: config => config.GroupInstanceId = $"greeter-{Guid.NewGuid()}" +``` + +**Infrastructure validation caveat**: + +With the consumer protocol, group membership is broker-driven and completes asynchronously: subscribing to a non-existent topic does not fail synchronously at consumer creation. Infrastructure validation via `OnMissingChannel.Validate` therefore has weaker guarantees than with the classic protocol and cannot reliably detect missing topics. Prefer `OnMissingChannel.Create`, or provision topics out-of-band (for example, via IaC). + +**Deprecated subscription properties**: + +`KafkaSubscription.SessionTimeout` and `KafkaSubscription.PartitionAssignmentStrategy` are now marked obsolete and will be removed in a future version. Their values are still honored as back-fill defaults for the classic protocol, but new code should configure these values on a `ClassicGroupProtocol` instead. + +## Offset Management + +It is important to understand how Brighter manages the **offset** of any **partitions** assigned to your **consumer**. + +- Brighter manages committing **offsets** to Kafka. This means we set the Confluent client's *auto store* and *auto commit* properties to *false*. +- The **CommitBatchSize** setting on the *Subscription* determines the size of your buffer. A smaller buffer is less efficient, but if your consumer crashes any **offsets** pending commit in the buffer will be lost, and you will be represented with those records when you next read from the **partition**. We default this value to 10. +- We do not add an offset commit to the buffer until you Ack the request. The message pump will Ack for you once you exit your handler (via return or [throwing an exception](/contents/HandlerFailure.md)). +- Flushing the commit buffer happens on a separate thread. We only run one flush at a time, and we flush a **CommitBatchSize** number of items from the buffer. + - A busy consumer may not flush on every increment of the **CommitBatchSize**, as it may need to wait for the last flush to finish. + - We won't flush again until we cross the next multiple of the **CommitBatchSize**. For example if the **CommitBatchSize** is 10, and the handler is busy so that by the time the buffer flushes there are 13 pending commits in the buffer, the buffer would only flush 10, and 3 would remain in the buffer; we would not flush the next 10 until the buffer hit 20. + - If your **CommitBatchSize** is too low for the throughput, you might find that you miss a flush interval, because you are already flushing. + - If you miss a flush on a busy consumer, your buffer will begin to back up. If this continues, you will not catch up with subsequent flushes, which only flush the **CommitBatchSize** each time. This would lead to you continually being "backed up". + - For this reason you must set a **CommitBatchSize** that keeps pace with the throughput of your consumer. Use a larger **CommitBatchSize** for higher throughput consumers, smaller for lower. +- We sweep uncommitted offsets at an interval. This triggers a flush if no flush has run since the last flush plus the *Subscription's* **SweepUncommittedOffsetsIntervalMs**. + - A sweep will not run if a flush is currently running (and will in turn block a flush). + - A sweep flushes a **CommitBatchSize** worth of commits. + - It is intended for low-throughput consumers where commits might otherwise languish waiting for a batch-size increment. + - It is *not* intended to flush a buffer that backs up because the **CommitBatchSize** is too low, and won't function for that. Fix the **CommitBatchSize** instead. +- On a re-balance where we stop processing a **partition** on an individual consumer, we flush the remaining **offsets** for the revoked **partitions**. + - We configure the consumer to use sticky assignment strategy to avoid unnecessary re-assignments (see the [Confluent documentation](https://www.confluent.io/blog/cooperative-rebalancing-in-kafka-streams-consumer-ksqldb/)). +- On a consumer shutdown we flush the buffer to commit all **offsets**. + +## Working with Schema Registry + +If you want to use tools within the Kafka ecosystem such as Kafka Connect or KQSL you will almost certainly need to use Confluent Schema Registry to provide the schema of your message. + +You will need to pull in the following package: + +* Confluent.SchemaRegistry + +and a package for the serialization of your choice. Here we are using JSON, so we use + +* Confluent.SchemaRegistry.Serdes.Json + +When working with Brighter, to use Confluent Schema Registry you will need to take a dependency on ISchemaRegistry in the constructor of your message mapper. To fulfill this constructor, in your application setup you will need to register an instance of schema registry. You should configure the schema registry config url to be the url of you schema registry. (Here we just use localhost for a development instance running in docker as an example). + +``` csharp +var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081"}; +var cachedSchemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryConfig); +services.AddSingleton(cachedSchemaRegistryClient); +``` + +Once you can satisfy the dependency, you will want to use the serializer from the Serdes package to serialize the body of your message, instead of System.Text.Json. Note that 'under-the-hood' the Serdes serializer uses [Json.NET](https://www.newtonsoft.com/json) and [NJsonSchema](https://github.com/RicoSuter/NJsonSchema), so you may need to mark up your code with attributes from these packages to create the schema you want and serialize a valid message to it. (Note that, at this time, the Serdes package does not support System.Text.Json so you will need to take a dependency on Json.NET if you want to use the schema registry). + +It is worth noting the following aspects of the code sample below: + +* We need to set up a SerializationContext and tell Serdes that we are serializing the message body using their serializer +* We provide two helpers, though you can pass your own settings if you prefer: + * **ConfluentJsonSerializationConfig.SerdesJsonSerializerConfig()** offers default settings for JSON serialization (many of these are passed through to Json.NET). + * **ConfluentJsonSerializationConfig.NJsonSchemaGeneratorSettings()** offers default settings for JSON Schema generation (such as using camelCase). + +``` csharp +public class GreetingEventMessageMapper : IAmAMessageMapper +{ + private readonly ISchemaRegistryClient _schemaRegistryClient; + private readonly string _partitionKey = "KafkaTestQueueExample_Partition_One"; + private SerializationContext _serializationContext; + private const string Topic = "greeting.event"; + + public GreetingEventMessageMapper(ISchemaRegistryClient schemaRegistryClient) + { + _schemaRegistryClient = schemaRegistryClient; + //We care about ensuring that we serialize the body using the Confluent tooling, as it registers and validates schema + _serializationContext = new SerializationContext(MessageComponentType.Value, Topic); + } + + public Message MapToMessage(GreetingEvent request) + { + var header = new MessageHeader(messageId: request.Id, topic: Topic, messageType: MessageType.MT_EVENT); + //This uses the Confluent JSON serializer, which wraps Newtonsoft but also performs schema registration and validation + var serializer = new JsonSerializer(_schemaRegistryClient, ConfluentJsonSerializationConfig.SerdesJsonSerializerConfig(), ConfluentJsonSerializationConfig.NJsonSchemaGeneratorSettings()).AsSyncOverAsync(); + var s = serializer.Serialize(request, _serializationContext); + var body = new MessageBody(s, "JSON"); + header.PartitionKey = _partitionKey; + + var message = new Message(header, body); + return message; + } + + public GreetingEvent MapToRequest(Message message) + { + var deserializer = new JsonDeserializer().AsSyncOverAsync(); + //This uses the Confluent JSON serializer, which wraps Newtonsoft but also performs schema registration and validation + var greetingCommand = deserializer.Deserialize(message.Body.Bytes, message.Body.Bytes is null, _serializationContext); + + return greetingCommand; + } +} + +``` + +## Requeue with Delay (Non-Blocking Retry) + +We don't currently support requeue with delay for Kafka. This is also known as non-blocking retry. With a stream if your app cannot process a record but it might be able to process the record after a delay (for example the DB is temporarily unavailable) then the options are: + +* Blocking Retry - keep retrying the processing of this record +* Load Shedding - ack the record to commit the offset, skipping this record +* Non-Blocking Retry - move the record to a new store or queue, skipping the original, append after a delay + +Brighter supports the first two of these options. + +* Blocking Retry - use a Polly policy via the **UsePolicy** attribute +* Load Shedding - allow the handler to complete, or throw an exception. This will cause the handler to commit the offset. + +Note that Blocking Retry means you will apply backpressure as the blocking retry means you will pause consumption until the record can be processed. + +A non-blocking retry typically creates a copy of the current record, and appends it to the stream so that it can be processed later: + +- Publish the message to be requeued to a new stream or store with a timestamp +- Ack the existing message so at to commit the offset +- Poll that stream or store and publish anything whose timestamp + delay means it is now due + +(You may need multiple tables or streams to support different delay lengths) + +Until Brighter supports this for you, implementation of non-blocking consumers is left to the user. From 4e9f49f1400e67925c368f785a3d7e35e3d8207e Mon Sep 17 00:00:00 2001 From: Rafael Lillo <7280959+lillo42@users.noreply.github.com> Date: Thu, 23 Jul 2026 20:53:04 +0100 Subject: [PATCH 3/3] V10MigrationGuide: add Kafka KIP-848 consumer group protocol as a V10 feature Document the new GroupProtocol option, the obsoleted SessionTimeout and PartitionAssignmentStrategy properties (with before/after examples), and how to opt into the KIP-848 consumer protocol. Refs BrighterCommand/Brighter#4233 --- contents/V10MigrationGuide.md | 856 +++++++++++++++++++++++++++++++++- 1 file changed, 855 insertions(+), 1 deletion(-) diff --git a/contents/V10MigrationGuide.md b/contents/V10MigrationGuide.md index 5c3d92d..318f98a 100644 --- a/contents/V10MigrationGuide.md +++ b/contents/V10MigrationGuide.md @@ -1 +1,855 @@ -__V10__ \ No newline at end of file +# Brighter V10 Migration Guide + +## Overview + +Brighter V10 introduces significant improvements and new features while maintaining a clear migration path from V9. This guide provides step-by-step instructions for upgrading your application to V10, addressing breaking changes, and adopting new features. + +**Key Changes in V10**: + +- Cloud Events support +- OpenTelemetry Semantic Conventions +- Default Message Mappers +- Dynamic Message Deserialization +- Nullable Reference Types (breaking) +- Simplified Configuration (breaking) +- Reactor and Proactor terminology (breaking) +- Polly Resilience Pipeline v8 (breaking) +- Request Context enhancements +- InMemory options for testing +- Transport improvements (PostgreSQL, RabbitMQ, Kafka, AWS) +- Request ID is now an `Id` type + +**Migration Effort**: Most applications can be migrated in **1-4 hours**, depending on complexity. + +## Before You Start + +### Prerequisites + +1. **Backup your code**: Commit all changes to version control +2. **Review the release notes**: Read [Release Notes](https://github.com/BrighterCommand/Brighter/releases) for V10 +3. **Update test suite**: Ensure your tests are passing on V9 +4. **Check dependencies**: Review third-party package compatibility + +### Recommended Migration Approach + +1. **Upgrade in a feature branch**: Don't upgrade directly in main/master +2. **Address breaking changes first**: Fix compilation errors before adopting new features +3. **Test thoroughly**: Run your full test suite after each step +4. **Deploy to staging**: Validate in a non-production environment +5. **Monitor production**: Watch for issues after deployment + +## Step 1: Update Package References + +### Update NuGet Packages + +Update all Brighter packages to V10: + +```bash +# Core packages +dotnet add package Paramore.Brighter --version 10.0.0 +dotnet add package Paramore.Brighter.Extensions.DependencyInjection --version 10.0.0 + +# Transport packages (update as needed) +dotnet add package Paramore.Brighter.MessagingGateway.RMQ --version 10.0.0 +dotnet add package Paramore.Brighter.MessagingGateway.Kafka --version 10.0.0 +dotnet add package Paramore.Brighter.MessagingGateway.AWSSQS --version 10.0.0 + +# Outbox/Inbox packages (update as needed) +dotnet add package Paramore.Brighter.Outbox.MsSql --version 10.0.0 +dotnet add package Paramore.Brighter.Inbox.MsSql --version 10.0.0 +``` + +### Check for Package Conflicts + +```bash +# List all packages and check for conflicts +dotnet list package +``` + +## Step 2: Address Breaking Changes + +### 1. Nullable Reference Types + +**Breaking Change**: Nullable reference types are now enabled across all Brighter projects. + +**Migration Steps**: + +1. **Enable nullable reference types** in your project (if not already enabled): + +```xml + + + enable + + +``` + +2. **Address compiler warnings** in Commands, Events, and Handlers: + +**Before (V9)**: + +```csharp +public class CreatePersonCommand : Command +{ + public string Name { get; set; } // Warning: Non-nullable property + public string Email { get; set; } // Warning: Non-nullable property +} +``` + +**After (V10)**: + +```csharp +public class CreatePersonCommand : Command +{ + public required string Name { get; set; } // Required property + public required string Email { get; set; } // Required property + + // Or with constructor + public CreatePersonCommand(Guid id, string name, string email) : base(id) + { + Name = name; + Email = email; + } +} +``` + +3. **Update Message Mappers** to handle nullable warnings: + +```csharp +public class PersonCreatedMapper : IAmAMessageMapper +{ + public Message MapToMessage(PersonCreated request) + { + // Validate non-null properties + ArgumentNullException.ThrowIfNull(request.Name); + + var header = new MessageHeader( + messageId: request.Id, + topic: new RoutingKey("PersonCreated"), + messageType: MessageType.MT_EVENT + ); + + var body = new MessageBody(JsonSerializer.Serialize(request)); + return new Message(header, body); + } +} +``` + +**See also**: [Nullable Reference Types Documentation](/contents/NullableReferenceTypes.md) + +### 2. Simplified Configuration + +**Breaking Change**: Builder methods renamed for clarity. + +**Before (V9)**: +```csharp +services.AddBrighter() + .UseExternalBus(new RmqProducerRegistryFactory(...).Create()) + .AddServiceActivator(options => + { + options.Subscriptions = subscriptions; + options.ChannelFactory = new ChannelFactory(...); + }); +``` + +**After (V10)**: +```csharp +services.AddBrighter() + .AddProducers(options => + { + options.ProducerRegistry = new RmqProducerRegistryFactory(...).Create(); + }) + .AddConsumers(options => + { + options.Subscriptions = subscriptions; + options.ChannelFactory = new ChannelFactory(...); + }); +``` + +**Migration Steps**: + +1. Replace `UseExternalBus` with `AddProducers` +2. Replace `AddServiceActivator` with `AddConsumers` +3. Update property names: `ProducerRegistry` instead of passing directly + +### 3. Reactor and Proactor Terminology + +**Breaking Change**: The `runAsync` flag on Subscription has been renamed to `MessagePumpType`. + +**Before (V9)**: + +```csharp +var subscription = new Subscription( + new SubscriptionName("my-subscription"), + new ChannelName("my-channel"), + new RoutingKey("my.routing.key"), + runAsync: true // Old parameter +); +``` + +**After (V10)**: + +```csharp +var subscription = new Subscription( + new SubscriptionName("my-subscription"), + new ChannelName("my-channel"), + new RoutingKey("my.routing.key"), + messagePumpType: MessagePumpType.Proactor // New parameter +); +``` + +**Migration Table**: + +| V9 Parameter | V10 Parameter | Description | +|--------------|---------------|-------------| +| `runAsync: false` | `messagePumpType: MessagePumpType.Reactor` | Synchronous, blocking I/O | +| `runAsync: true` | `messagePumpType: MessagePumpType.Proactor` | Asynchronous, non-blocking I/O | + +**See also**: [Reactor and Proactor Documentation](/contents/ReactorAndProactor.md) + +### 4. Polly Resilience Pipeline + +**Breaking Change**: `TimeoutPolicyAttribute` is obsolete. Use `UseResiliencePipeline` attribute. + +**Before (V9)**: + +```csharp +public class MyHandler : RequestHandlerAsync +{ + [TimeoutPolicy(milliseconds: 5000, step: 1)] + public override async Task HandleAsync( + MyCommand command, + CancellationToken cancellationToken = default) + { + // Handler logic + return await base.HandleAsync(command, cancellationToken); + } +} +``` + +**After (V10)**: + +1. **Define a Resilience Pipeline**: + +```csharp +var resiliencePipelineRegistry = new ResiliencePipelineRegistry(); +resiliencePipelineRegistry.TryAddBuilder>( + "MyPipeline", + (builder, context) => + { + builder.AddTimeout(TimeSpan.FromSeconds(5)); + builder.AddRetry(new RetryStrategyOptions + { + MaxRetryAttempts = 3, + Delay = TimeSpan.FromMilliseconds(100) + }); + }); +``` + +2. **Use the new attribute**: + +```csharp +public class MyHandler : RequestHandlerAsync +{ + [UseResiliencePipeline(policy: "MyPipeline", step: 1)] + public override async Task HandleAsync( + MyCommand command, + CancellationToken cancellationToken = default) + { + // Handler logic + return await base.HandleAsync(command, cancellationToken); + } +} +``` + +3. **Register the pipeline** with Brighter: + +```csharp +services.AddBrighter(options => +{ + options.PolicyRegistry = resiliencePipelineRegistry; +}); +``` + +**See also**: [Policy Retry and Circuit Breaker Documentation](/contents/PolicyRetryAndCircuitBreaker.md) + +### 5. Request Context Interface Changes + +**Breaking Change**: `IRequestContext` interface has new properties. + +**New Properties**: + +- `PartitionKey`: Set message partition keys dynamically +- `CustomHeaders`: Add custom headers via request context +- `ResilienceContext`: Integration with Polly resilience pipeline +- `OriginatingMessage`: Access the original message (for consumers) + +**Migration**: Most code should not be affected unless you implement `IRequestContext` directly. + +**If you implement `IRequestContext`** (rare): + +```csharp +public class MyRequestContext : IRequestContext +{ + public Guid Id { get; set; } + public ISpan Span { get; set; } + public Dictionary Bag { get; set; } + + // V10: Add new properties + public string? PartitionKey { get; set; } + public Dictionary CustomHeaders { get; set; } = new(); + public ResilienceContext? ResilienceContext { get; set; } + public Message? OriginatingMessage { get; set; } +} +``` + +**Using new properties**: + +```csharp +public class MyHandler : RequestHandlerAsync +{ + public override async Task HandleAsync( + MyCommand command, + CancellationToken cancellationToken = default) + { + // Set partition key for message routing + Context.PartitionKey = command.TenantId; + + // Add custom headers + Context.CustomHeaders["X-Correlation-Id"] = command.CorrelationId; + + // Access originating message (for consumers) + if (Context.OriginatingMessage != null) + { + var receivedTimestamp = Context.OriginatingMessage.Header.TimeStamp; + } + + return await base.HandleAsync(command, cancellationToken); + } +} +``` + +### 6. Request Id and CorrelationId type change + +In V9 the `Request` type used a `Guid` to represent the identity of a `Command` or `Event`. In V10, as part of a move away from primitives, we have changed this to be a type `Id`. An `Id` can be constructed from a `string` using its `ToString()` method. If you have used a `Guid` then you will need to turn the `Guid` into a `string`. + +Within `IRequest` both `Id` and `CorrelationId` both use the type `Id` in V10. + +If you were using a `Guid` to create a random identity, you can just use `Id.Random()` instead which has the same behavior. + +Before: + +```csharp + +class MyCommand() : Command(Guid.NewGuid()) +{ + public string Value { get; set; } = string.Empty; +} + +``` + +After: + +```csharp + +class MyCommand() : Command(Id.Random()) +{ + public string Value { get; set; } = string.Empty; +} + +``` + +## Step 3: Adopt New Features (Optional) + +### 1. Cloud Events Support + +V10 adds full Cloud Events specification support. + +**Benefits**: + +- Standardized event metadata +- Better interoperability with other systems +- Rich event context information + +**Adoption Steps**: + +1. **Update Publication** to include Cloud Events properties: + +```csharp +new RmqPublication +{ + Topic = new RoutingKey("PersonCreated"), + CloudEventsType = new CloudEventsType("io.paramore.person.created"), + Source = new Uri("https://api.example.com/persons"), + Subject = "person/created", + MakeChannels = OnMissingChannel.Create +} +``` + +2. **Use Cloud Events headers** in your mapper (optional): + +```csharp +public class PersonCreatedMapper : IAmAMessageMapper +{ + public Message MapToMessage(PersonCreated request, Publication publication) + { + var header = new MessageHeader( + messageId: request.Id, + topic: publication.Topic, + messageType: MessageType.MT_EVENT, + type: publication.CloudEventsType, // Use Cloud Events type + source: publication.Source, + subject: publication.Subject + ); + + var body = new MessageBody(JsonSerializer.Serialize(request)); + return new Message(header, body); + } +} +``` + +**See also**: [Cloud Events Documentation](/contents/CloudEventsSupport.md) + +### 2. Default Message Mappers + +V10 allows you to omit message mappers for simple JSON serialization. + +**Benefits**: + +- Less boilerplate code +- Faster development +- Still supports custom mappers for complex scenarios + +**Before (V9) - Required Mapper**: + +```csharp +public class PersonCreatedMapper : IAmAMessageMapper +{ + public Message MapToMessage(PersonCreated request) + { + var header = new MessageHeader( + messageId: request.Id, + topic: new RoutingKey("PersonCreated"), + messageType: MessageType.MT_EVENT + ); + + var body = new MessageBody(JsonSerializer.Serialize(request)); + return new Message(header, body); + } + + public PersonCreated MapToRequest(Message message) + { + return JsonSerializer.Deserialize(message.Body.Value) + ?? throw new InvalidOperationException("Failed to deserialize"); + } +} + +// Register mapper +messageMapperRegistry.Register(); +``` + +**After (V10) - No Mapper Needed**: + +```csharp +// No mapper registration needed for simple JSON serialization! +// Brighter uses JsonMessageMapper by default + +// Just define your message +public class PersonCreated : Event +{ + public string Name { get; set; } + public string Email { get; set; } +} + +// Publish directly +await commandProcessor.PublishAsync(new PersonCreated +{ + Name = "Alice", + Email = "alice@example.com" +}); +``` + +**When to use custom mappers**: + +- Complex transformations +- Non-JSON formats (XML, Protobuf, etc.) +- Transform pipelines (encryption, compression, claim check) +- Custom header mapping + +**See also**: [Message Mappers Documentation](/contents/MessageMappers.md) + +### 3. Dynamic Message Deserialization + +V10 supports multiple message types on the same channel. + +**Benefits**: + +- Content-based routing +- Flexible message processing +- Better use of infrastructure + +**Example**: + +```csharp +new KafkaSubscription( + new SubscriptionName("task-state-subscription"), + channelName: new ChannelName("task.state"), + routingKey: new RoutingKey("task.update"), + getRequestType: message => message switch + { + var m when m.Header.Type == new CloudEventsType("io.paramore.task.created") + => typeof(TaskCreated), + var m when m.Header.Type == new CloudEventsType("io.paramore.task.updated") + => typeof(TaskUpdated), + var m when m.Header.Type == new CloudEventsType("io.paramore.task.deleted") + => typeof(TaskDeleted), + _ => throw new ArgumentException( + $"No type mapping found for message with type {message.Header.Type}", + nameof(message)) + }, + groupId: "task-consumer-group", + messagePumpType: MessagePumpType.Proactor +); +``` + +**See also**: [Dynamic Deserialization Documentation](/contents/DynamicDeserialization.md) + +### 4. OpenTelemetry Semantic Conventions + +V10 adopts OpenTelemetry Semantic Conventions for messaging. + +**Impact**: Trace spans will have different names and attributes than V9. + +**Benefits**: + +- Standard messaging conventions +- Better integration with APM tools +- Consistent telemetry across systems + +**Migration**: + +1. **Update dashboards and alerts** to use new span names +2. **Review trace queries** for compatibility +3. **Test observability** in staging environment + +**V9 Span Names**: + +- `Paramore.Brighter.CommandProcessor.Send` +- `Paramore.Brighter.MessagePump.Receive` + +**V10 Span Names** (OTel Semantic Conventions): + +- `messaging.send` +- `messaging.receive` +- `messaging.process` + +**See also**: [Telemetry Documentation](/contents/Telemetry.md) + +### 5. InMemory Options for Testing + +V10 provides comprehensive InMemory implementations for testing. + +**Benefits**: + +- Fast test execution +- No external dependencies +- Easy CI/CD integration + +**Example**: + +```csharp +// Test setup with InMemory components +var internalBus = new InternalBus(); + +services.AddBrighter(options => +{ + options.HandlerLifetime = ServiceLifetime.Scoped; +}) +.AddProducers(options => +{ + var publication = new Publication { Topic = new RoutingKey("TestTopic") }; + options.ProducerRegistry = new InMemoryProducerRegistryFactory( + internalBus, + new[] { publication }, + InstrumentationOptions.All + ).Create(); + options.Outbox = new InMemoryOutbox(TimeProvider.System); +}) +.AddConsumers(options => +{ + options.Inbox = new InboxConfiguration( + new InMemoryInbox(TimeProvider.System), + InboxConfiguration.NoActionOnExists + ); + options.Subscriptions = subscriptions; + options.ChannelFactory = new InMemoryChannelFactory(internalBus, TimeProvider.System); +}) +.UseScheduler(new InMemorySchedulerFactory()) +.AutoFromAssemblies(); +``` + +**See also**: [InMemory Options Documentation](/contents/InMemoryOptions.md) + +### 6. Kafka KIP-848 Consumer Group Protocol + +V10 adds support for [KIP-848](https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol), Kafka's next-generation consumer group protocol with broker-driven partition assignment (requires Apache Kafka 4.0 or later). + +**Not a breaking change**: Existing subscriptions continue to use the classic consumer group protocol by default. + +**Obsoleted properties**: `KafkaSubscription.SessionTimeout` and `KafkaSubscription.PartitionAssignmentStrategy` are now marked `[Obsolete]` and will be removed in a future version. They still work (their values are used as back-fill defaults), but new code should set these values on a `ClassicGroupProtocol`: + +**Before**: + +```csharp +var subscription = new KafkaSubscription( + subscriptionName: new SubscriptionName("paramore.example.greeting"), + channelName: new ChannelName("greeting.event"), + routingKey: new RoutingKey("greeting.event"), + groupId: "my-consumer-group", + sessionTimeout: TimeSpan.FromSeconds(30), + partitionAssignmentStrategy: PartitionAssignmentStrategy.Range +); +``` + +**After**: + +```csharp +var subscription = new KafkaSubscription( + subscriptionName: new SubscriptionName("paramore.example.greeting"), + channelName: new ChannelName("greeting.event"), + routingKey: new RoutingKey("greeting.event"), + groupId: "my-consumer-group" +) +{ + GroupProtocol = new ClassicGroupProtocol + { + SessionTimeout = TimeSpan.FromSeconds(30), + PartitionAssignmentStrategy = PartitionAssignmentStrategy.Range + } +}; +``` + +**Opting into KIP-848**: + +```csharp +var subscription = new KafkaSubscription( + subscriptionName: new SubscriptionName("paramore.example.greeting"), + channelName: new ChannelName("greeting.event"), + routingKey: new RoutingKey("greeting.event"), + groupId: "my-consumer-group" +) +{ + GroupProtocol = new ConsumerGroupProtocol() +}; +``` + +**See also**: [Kafka Configuration — Consumer Group Protocol (KIP-848)](/contents/KafkaConfiguration.md#consumer-group-protocol-kip-848) + +## Step 4: Test Your Migration + +### Unit Tests + +1. **Run existing unit tests**: + +```bash +dotnet test +``` + +2. **Address test failures**: + - Update mocks for `IRequestContext` new properties + - Update assertions for OpenTelemetry span names + - Fix nullable reference warnings + +### Integration Tests + +1. **Test with InMemory components** (fast): + +```csharp +[Fact] +public async Task Should_Process_Message_With_V10_Components() +{ + // Arrange + var internalBus = new InternalBus(); + var serviceProvider = BuildServiceProvider(internalBus); + var commandProcessor = serviceProvider.GetRequiredService(); + + // Act + await commandProcessor.PublishAsync(new PersonCreated + { + Name = "Alice", + Email = "alice@example.com" + }); + + // Assert + var messages = internalBus.Stream(new RoutingKey("PersonCreated")); + Assert.NotEmpty(messages); +} +``` + +2. **Test with real transports** (slower, more complete): + +```bash +# Start dependencies (Docker Compose) +docker-compose up -d rabbitmq postgres + +# Run integration tests +dotnet test --filter Category=Integration +``` + +### Performance Testing + +Compare V9 and V10 performance: + +```csharp +[Benchmark] +public async Task V10_MessageProcessing() +{ + for (int i = 0; i < 1000; i++) + { + await _commandProcessor.SendAsync(new TestCommand { Value = i }); + } +} +``` + +Expected: V10 should have similar or better performance due to optimizations. + +## Step 5: Deploy to Staging + +### Pre-Deployment Checklist + +- [ ] All tests passing +- [ ] Code review completed +- [ ] Documentation updated +- [ ] Release notes reviewed +- [ ] Rollback plan prepared + +### Deployment Steps + +1. **Deploy to staging environment** +2. **Run smoke tests** +3. **Monitor metrics**: + - Message processing latency + - Error rates + - Resource usage (CPU, memory) +4. **Check logs** for warnings or errors +5. **Validate telemetry** (traces, metrics) + +### Monitoring + +Watch for: +- Increased error rates +- Performance degradation +- OpenTelemetry trace issues +- Null reference exceptions + +## Step 6: Deploy to Production + +### Production Deployment + +1. **Deploy during low-traffic period** +2. **Use blue-green or canary deployment** if possible +3. **Monitor closely** for first 24 hours +4. **Have rollback plan ready** + +### Post-Deployment + +- Review logs for issues +- Check application metrics +- Validate message processing +- Monitor OpenTelemetry traces + +## Common Migration Issues + +### Issue 1: Nullable Reference Warnings + +**Symptom**: CS8618, CS8600, CS8602 warnings + +**Solution**: See [Nullable Reference Types Documentation](/contents/NullableReferenceTypes.md) + +### Issue 2: Method Not Found + +**Symptom**: `UseExternalBus` method not found + +**Solution**: Replace with `AddProducers` + +### Issue 3: Property Not Found on Subscription + +**Symptom**: `runAsync` property does not exist + +**Solution**: Use `messagePumpType: MessagePumpType.Proactor` or `MessagePumpType.Reactor` + +### Issue 4: TimeoutPolicy Not Working + +**Symptom**: `TimeoutPolicyAttribute` marked as obsolete + +**Solution**: Migrate to `UseResiliencePipeline` with Polly v8 + +### Issue 5: Telemetry Spans Changed + +**Symptom**: Dashboard queries not finding spans + +**Solution**: Update queries to use OpenTelemetry Semantic Convention names + +## Rollback Plan + +If you need to roll back to V9: + +1. **Revert package versions**: + +```bash +dotnet add package Paramore.Brighter --version 9.x.x +``` + +2. **Revert code changes**: + +```bash +git revert +``` + +3. **Redeploy V9 version** + +4. **Investigate issues** before attempting V10 migration again + +## Getting Help + +### Resources + +- [Brighter Documentation](https://brightercommand.github.io/Brighter/) +- [GitHub Issues](https://github.com/BrighterCommand/Brighter/issues) +- [Release Notes](https://github.com/BrighterCommand/Brighter/releases) +- [Stack Overflow](https://stackoverflow.com/questions/tagged/brighter) + +### Reporting Issues + +If you encounter issues during migration: + +1. **Check existing issues**: Search [GitHub Issues](https://github.com/BrighterCommand/Brighter/issues) +2. **Create a new issue** with: + - V9 and V10 versions + - Minimal reproduction example + - Error messages and stack traces + - Environment details (.NET version, OS, transport) + +## Summary + +Migrating to Brighter V10 involves: + +1. **Update packages** to V10 +2. **Address breaking changes**: + - Nullable reference types + - Simplified configuration API + - Reactor/Proactor terminology + - Polly Resilience Pipeline +3. **Adopt new features** (optional): + - Cloud Events + - Default message mappers + - Dynamic deserialization + - OpenTelemetry conventions + - InMemory testing options +4. **Test thoroughly** +5. **Deploy to staging**, then production +6. **Monitor** and address any issues + +Most migrations can be completed in **1-4 hours**. The breaking changes are straightforward, and V10 provides significant improvements in features, performance, and developer experience. + +Good luck with your migration! 🚀