Option to propagate the Kafka partition key to the Wolverine Envelope.GroupId

[lyall-sc]

Is your feature request related to a problem? Please describe.
The problem is our requirement to set the Kafka partition key to be the group/stream id without manually doing this for every message type. We currently solve this problem by inspecting all message types that are configured for Kafka publishing during bootstrapping. We check that they all have a string Id property, and then when returning a cascaded message we pass all message objects through a class that has a map of "valid publishable message types" and extract the Id property value, and set it as the partition key.

Describe the solution you'd like
It would be great if we could bin this code and have a Wolverine config option to automatically assign the group/stream id to be the Kafka partition key. We don't have marker interfaces on our message types since we auto-gen them from .proto files.

Describe alternatives you've considered
We have a working alternative as described but would be delighted with a more elegant inbuilt solution.

[jeremydmiller]

@lyall-sc To be more specific, Wolverine can already happily do this for the "aggregate handler workflow". Your issue was needing this for messages that created new streams, correct?

[lyall-sc]

Hmm no. This is when we're publishing messages, e.g.

public static class CreateStuffCommandHandler
{
    [AggregateHandler]
    public static (Events, OutgoingMessages) Handle(CreateStuffCommand msg, StuffAggregate aggregate)
    {
        aggregate ??= new StuffAggregate();

        // decider logic
        var evt = new StuffCreatedEvent(msg.Id, msg.Stuff);

        return ([evt], [evt]); // here
    }
}

At "here", we want to set the PartitionKey automatically to be the stream id. As it stands, we inject another custom type, which has an "Add" method which does OutgoingMessages.Add(o, new DeliveryOptions { PartitionKey = id }); where id is the value of the "Id" property on the message object (which we know is there by building a map of publishable message types during bootstrapping), and we all know to use that type's "Add" method when creating new events, and then return its value for OutgoingMessages in the tuple.

Is there a way to have exactly the code I have above, with none of the other nonsense I just described, and have it automatically set the PartitionKey to be the stream id?

[jeremydmiller]

Any reason why the built in inferred group id rule isn't working here? Wolverine would be able to infer the stream id from the CreateStuffCommand above.

Ah, you're wanting the outgoing messages to automatically carry forward the stream id as the partition key on the outgoing messages? My understanding was way off before then

[jeremydmiller]

Approach

• Create a new IEnvelopeRule that propagates the MessageContext.Envelope.GroupId to the Envelope.PartitionKey of the outgoing envelope
• Add a unit test for that behavior
• Add a new WolverineOptions.MetadataRules of type List that's internal
• Add a new method to IPolicies that adds the new group id to partition key rule to the WolverineOptions.MetadataRules
• Inside of MessageBus.PersistOrSendAsync, loop through all the MetdataRules and call the ApplyCorrelation() method on the outgoing envelope
• In the Kafka tests, add a test that uses this new group id to partition key and verify that the messages received by that Kafka topic have the partition key set to the originating messages's group id. Utilize the partitioned sequential messaging tests for Kafka
• Add a note about this new rule in the documentation for Kafka and for the partitioned sequential messaging documentation with a new code example
• Run both the CoreTests and the Kafka test suite to verify the work