Document Kafka partitioners, analyzers, and code fixes

SUMMARY.md

@@ -12,6 +12,7 @@
 * [InMemory Options for Development and Testing](/contents/InMemoryOptions.md)
 * [Test Double Options for Command Processor](/contents/TestDoubleOptions.md)
 * [Pipeline Validation and Diagnostics](/contents/PipelineValidation.md)
+* [Analyzer Support](/contents/AnalyzerSupport.md)
 
 ## Darker Configuration
 

contents/AnalyzerSupport.md

@@ -0,0 +1,181 @@
+# Analyzer Support
+
+Brighter provides Roslyn analyzers that detect common configuration and message-mapping mistakes while you write and build your application. The analyzers surface these problems as IDE and compiler warnings, before they can become runtime errors or subtle production behavior.
+
+The Brighter analyzer package also includes code fixes for supported diagnostics. A code fix lets your IDE apply the recommended change through Quick Actions instead of editing the code manually.
+
+## Installing the Analyzer
+
+Add the Brighter analyzer NuGet package to each project that creates Brighter publications, subscriptions, or message mappers:
+
+```shell
+dotnet add package Paramore.Brighter.Analyzer.Package
+```
+
+The analyzer and code-fix assemblies load automatically for the project. You do not need to register the analyzer in your Brighter configuration.
+
+## Diagnostic Reference
+
+| ID | Severity | Detects | Code fix |
+| --- | --- | --- | --- |
+| **BRT001** | Warning | A `Publication` is created without assigning `RequestType`. | No |
+| **BRT002** | Warning | The type assigned to `RequestType` does not implement `IRequest`. | No |
+| **BRT003** | Warning | A `Subscription` is created without specifying `MessagePumpType`. | No |
+| **BRT004** | Warning | A wrap attribute is applied to the wrong message-mapper method. | No |
+| **BRT005** | Warning | An unwrap attribute is applied to the wrong message-mapper method. | No |
+| **BRT006** | Warning | A `KafkaPublication` is created without an explicit `Partitioner` assignment. | Yes |
+| **BRT007** | Warning | A `KafkaPublication` uses `Partitioner.ConsistentRandom`. | Yes |
+| **BRT008** | Warning | A `KafkaPublication` uses `Partitioner.Consistent`. | Yes |
+
+## Kafka Partitioner Diagnostics
+
+The Kafka partitioner analyzer checks `KafkaPublication` and `KafkaPublication<T>` object creations. It helps you make an explicit partitioner choice and recommends the Murmur2-based partitioners for new publications.
+
+When you do not assign `Partitioner`, `KafkaPublication` currently defaults to `Partitioner.ConsistentRandom`. That default preserves compatibility, but it also hides an important partitioning decision. The partitioner controls how message keys map to Kafka partitions; an uneven mapping can create *hot partitions*, where a small number of partitions and consumers receive a disproportionate share of the work.
+
+For new publications, prefer `Partitioner.Murmur2Random`. It uses the Murmur2 hash for keyed messages and spreads unkeyed messages randomly across partitions. Use `Partitioner.Murmur2` when you do not expect unkeyed messages and want those messages to use a single deterministic partition.
+
+### BRT006: Missing Partitioner
+
+**BRT006** warns when a `KafkaPublication` does not assign `Partitioner` explicitly:
+
+```csharp
+using Paramore.Brighter;
+using Paramore.Brighter.MessagingGateway.Kafka;
+
+var publication = new KafkaPublication
+{
+    Topic = new RoutingKey("orders.created")
+    // Warning: Partitioner assignment is missing.
+};
+```
+
+Set the partitioner explicitly:
+
+```csharp
+using Paramore.Brighter;
+using Paramore.Brighter.MessagingGateway.Kafka;
+
+var publication = new KafkaPublication
+{
+    Topic = new RoutingKey("orders.created"),
+    Partitioner = Partitioner.Murmur2Random
+};
+```
+
+The code fix adds `Partitioner = Partitioner.Murmur2Random` to the publication initializer.
+
+### BRT007: ConsistentRandom Partitioner Used
+
+**BRT007** warns when a publication uses `Partitioner.ConsistentRandom`:
+
+```csharp
+var publication = new KafkaPublication
+{
+    Topic = new RoutingKey("orders.created"),
+    Partitioner = Partitioner.ConsistentRandom // Warning: prefer Murmur2Random.
+};
+```
+
+For a new publication, change the value to `Murmur2Random`:
+
+```csharp
+var publication = new KafkaPublication
+{
+    Topic = new RoutingKey("orders.created"),
+    Partitioner = Partitioner.Murmur2Random
+};
+```
+
+The code fix replaces `Partitioner.ConsistentRandom` with `Partitioner.Murmur2Random`.
+
+### BRT008: Consistent Partitioner Used
+
+**BRT008** warns when a publication uses `Partitioner.Consistent`:
+
+```csharp
+var publication = new KafkaPublication
+{
+    Topic = new RoutingKey("orders.created"),
+    Partitioner = Partitioner.Consistent // Warning: prefer Murmur2.
+};
+```
+
+For a new publication, change the value to `Murmur2`:
+
+```csharp
+var publication = new KafkaPublication
+{
+    Topic = new RoutingKey("orders.created"),
+    Partitioner = Partitioner.Murmur2
+};
+```
+
+The code fix replaces `Partitioner.Consistent` with `Partitioner.Murmur2`.
+
+## Applying Code Fixes
+
+The analyzer package includes code fixes for the Kafka partitioner diagnostics:
+
+| Diagnostic | Quick Action |
+| --- | --- |
+| **BRT006** | Set `Partitioner` to `Partitioner.Murmur2Random` |
+| **BRT007** | Use `Partitioner.Murmur2Random` |
+| **BRT008** | Use `Partitioner.Murmur2` |
+
+To apply a fix:
+
+1. Place the caret on the warning in your IDE.
+2. Open Quick Actions, usually with **Ctrl+.** or the light-bulb icon.
+3. Select the recommended partitioner action.
+4. Review the change before saving.
+
+The code-fix providers support batch fixing, so IDEs that expose Roslyn **Fix All** operations can apply the same fix across a document, project, or solution.
+
+If the file does not already import the Kafka namespace, make sure the fixed code can resolve the `Partitioner` enum:
+
+```csharp
+using Paramore.Brighter.MessagingGateway.Kafka;
+```
+
+## Existing Kafka Topics
+
+Review partitioner warnings carefully before changing an existing topic. Different hash algorithms can map the same partition key to different partitions. Changing from `ConsistentRandom` to `Murmur2Random`, or from `Consistent` to `Murmur2`, can therefore move keys between partitions and affect per-key ordering during the transition.
+
+For an existing publication that must preserve its current key-to-partition mapping, you can keep the existing partitioner and suppress the warning locally.
+
+Use a pragma around a single publication:
+
+```csharp
+#pragma warning disable BRT007
+var publication = new KafkaPublication
+{
+    Topic = new RoutingKey("legacy.orders.created"),
+    Partitioner = Partitioner.ConsistentRandom // Intentional: preserve existing key mapping.
+};
+#pragma warning restore BRT007
+```
+
+Or configure the diagnostic in `.editorconfig`:
+
+```ini
+dotnet_diagnostic.BRT007.severity = none
+```
+
+Prefer a narrow suppression with an explanatory comment over disabling the diagnostic globally.
+
+## Best Practices
+
+- Set `Partitioner` explicitly on every `KafkaPublication`.
+- Use `Partitioner.Murmur2Random` for new publications unless you have a specific compatibility requirement.
+- Treat a partitioner change on an existing topic as a key-mapping change, not just a code cleanup.
+- Use **Fix All** only after checking that the publications in scope are safe to migrate.
+- Keep analyzer warnings enabled so new publications do not silently inherit the legacy default.
+
+## Further Reading
+
+- [Kafka Configuration: Kafka Hash Partitioning](/contents/KafkaConfiguration.md#kafka-hash-partitioning)
+- [Message Mappers](/contents/MessageMappers.md)
+- Reference code: `Brighter/src/Paramore.Brighter.Analyzer/Analyzers/KafkaPublicationPartitionerAnalyzer.cs`
+- Reference code: `Brighter/src/Paramore.Brighter.Analyzer.CodeFixes/CodeFixes/MissingPartitionerCodeFixProvider.cs`
+- Reference code: `Brighter/src/Paramore.Brighter.Analyzer.CodeFixes/CodeFixes/PartitionerValueCodeFixProvider.cs`

contents/KafkaConfiguration.md

@@ -11,7 +11,7 @@ Kafka has two main roles:
 
 **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**. 
+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. See [Kafka Hash Partitioning](#kafka-hash-partitioning) for how to control the hashing algorithm that Brighter uses to map a **partition key** to a **partition**. 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.
 
@@ -92,7 +92,7 @@ We allow you to configure properties for both Brighter and the Confluent .NET cl
 - **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.
+- **Partitioner**: How do we map a partition key to a partition? Defaults to Partitioner.ConsistentRandom, but we recommend Partitioner.Murmur2Random for a more even distribution of messages across partitions. See [Kafka Hash Partitioning](#kafka-hash-partitioning) below for the supported values and the differences between them.
 - **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.
@@ -123,6 +123,62 @@ The following example shows how a *Publication* might be configured:
 
 ```
 
+### Kafka Hash Partitioning
+
+A Kafka **topic** is split into **partitions**, and the producer decides which **partition** each message is written to. The algorithm that makes this decision is the **partitioner**, which Brighter exposes through the **Partitioner** property on a *Publication*. Brighter's **Partitioner** enum maps directly onto the Confluent .NET client's *partitioner* setting (from librdkafka).
+
+How the partitioner behaves depends on whether the message has a **partition key**. You set the partition key on the message header in your message mapper:
+
+``` csharp
+public Message MapToMessage(GreetingEvent request)
+{
+	var header = new MessageHeader(request.Id, "greeting.event", MessageType.MT_EVENT)
+	{
+		//Messages with the same partition key go to the same partition
+		PartitionKey = request.CustomerId.ToString()
+	};
+	...
+}
+```
+
+When a **partition key** is present, the partitioner hashes the key and selects a partition deterministically: all messages with the same key are written to the same **partition**, which preserves their order relative to one another (and means they are handled by the same consumer in a consumer group). When no key is set, the behavior depends on the partitioner variant, as described below.
+
+#### Supported Partitioners
+
+Brighter supports the following partitioners:
+
+| Partitioner | Keyed messages | Unkeyed messages | Notes |
+| --- | --- | --- | --- |
+| **Random** | Random partition | Random partition | Ignores the partition key entirely, so there are no per-key ordering guarantees. |
+| **Consistent** | CRC32 hash of the key | Always the same (single) partition | librdkafka's legacy consistent partitioner. CRC32 can cluster keys, risking uneven distribution. |
+| **ConsistentRandom** | CRC32 hash of the key | Random partition | Brighter's default. CRC32 can cluster keys, risking uneven distribution. |
+| **Murmur2** | Murmur2 hash of the key | Always the same (single) partition | Good key distribution, but the single partition for unkeyed messages can become a hot spot. |
+| **Murmur2Random** | Murmur2 hash of the key | Random partition | The most even distribution for both keyed and unkeyed messages. **Recommended.** |
+
+The difference between the *Consistent** family and the *Murmur2** family is the hash function used to map a key to a partition: **CRC32** versus **Murmur2**. Because the hash functions differ, the same key maps to a different partition under each family. The difference between each base variant and its **Random** counterpart is what happens to messages *without* a partition key: the base variants (**Consistent**, **Murmur2**) always write unkeyed messages to the same single partition, whereas the **Random** variants (**ConsistentRandom**, **Murmur2Random**) spread them randomly across partitions. A single partition for unkeyed messages can become a bottleneck and a hot spot, so the **Random** variants are generally preferable.
+
+#### Why We Recommend Murmur2Random
+
+We recommend setting **Partitioner** to **Partitioner.Murmur2Random** because of how it distributes messages across partitions:
+
+1. **More even distribution of keyed messages**: Murmur2 generally spreads keys more uniformly across partitions than the CRC32 hash used by **Consistent** and **ConsistentRandom**. This matters because uneven distribution creates "hot" partitions: a few partitions receive a disproportionate share of the messages, so the consumers assigned to those partitions become a bottleneck while the remaining consumers sit underused. Because only one consumer in a group may read a partition at a time, a hot partition caps your effective throughput and grows consumer lag no matter how many consumers you add.
+2. **No single partition for unkeyed messages**: where **Murmur2** (and **Consistent**) send every message without a partition key to the same partition—concentrating all of that load on one partition, and therefore on one consumer—**Murmur2Random** spreads unkeyed messages randomly across all partitions.
+
+``` csharp
+	new KafkaPublication[]
+	{
+		new KafkaPublication()
+		{
+			Topic = new RoutingKey("MyTopicName"),
+			NumPartitions = 3,
+			Partitioner = Partitioner.Murmur2Random,
+			MakeChannels = OnMissingChannel.Create
+		}
+	}
+```
+
+Note that changing the partitioner on an existing topic changes where keys land: messages with the same key may be written to different partitions before and after the change, which can break per-key ordering during the transition. Plan such a change for a deployment window where this is acceptable, or apply it when you create a new topic.
+
 ### 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<ProducerConfig> 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.
@@ -469,10 +525,3 @@ A non-blocking retry typically creates a copy of the current record, and appends
 (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.
-
-
-
-
-
-
-