feat(broker): Enables message keys for batch publishing

[ozeranskii]

Description

This PR adds support for per-message keys in Kafka/Confluent batch publishing operations. Previously, batch publishers didn't support keys at all. Now users can specify keys in three ways:

1. Per-message keys using KafkaPublishMessage (new alias for KafkaResponse)
2. Single key for entire batch via publish() parameter
3. Default key from publisher config (factory-level configuration)

Code Examples

1. Per-message keys using KafkaPublishMessage

from faststream.kafka import KafkaBroker, KafkaPublishMessage

broker = KafkaBroker()

@broker.subscriber("input")
async def handler():
    await broker.publish_batch(
        KafkaPublishMessage("user:1", key=b"user1"),
        KafkaPublishMessage("user:2", key=b"user2"),
        "user:3",  # Uses default key (None)
        topic="output"
    )

2. Single key for entire batch

publisher = broker.publisher("topic", batch=True)

await publisher.publish(
    "message1",
    "message2",
    "message3",
    key=b"partition_key"  # All messages get same key
)

3. Default key from factory config

publisher = broker.publisher(
    "topic",
    batch=True,
    key=b"default_key"  # Applied to all messages by default
)

await publisher.publish("msg1", "msg2")  # Both get default_key

# Override default for specific messages
await publisher.publish(
    KafkaPublishMessage("msg1", key=b"override"),
    "msg2"  # Uses default_key
)

Backward Compatibility

Fully backward compatible - all existing code continues to work:

• key parameter is optional (defaults to None)
• Existing batch publishers without keys work unchanged

Fixes #2514

Type of change

Please delete options that are not relevant.

• Documentation (typos, code examples, or any documentation updates)
• Bug fix (a non-breaking change that resolves an issue)
• New feature (a non-breaking change that adds functionality)
• Breaking change (a fix or feature that would disrupt existing functionality)
• This change requires a documentation update

Checklist

• My code adheres to the style guidelines of this project (just lint shows no errors)
• I have conducted a self-review of my own code
• I have made the necessary changes to the documentation
• My changes do not generate any new warnings
• I have added tests to validate the effectiveness of my fix or the functionality of my new feature
• Both new and existing unit tests pass successfully on my local environment by running just test-coverage
• I have ensured that static analysis tests are passing by running just static-analysis
• I have included code examples to illustrate the modifications

[Lancetnik]

Sorry, but magic {"message": ..., "key": ... } doesn't look as a good API.
One reason - we may want to add more additional information to each method instead of just key (headers, publish timestamp, etc)

This reason we have a feature already - Response (or KafkaResponse)

@broker.subscriber(...)
@broker.publisher(...)
async def handler():
     return Response("body", key=b"key")

This object allows to setup any information of outgoing message you want

So, I suggest to support smth like this Response object - Message (we can discuss naming)

await broker.publish_batch(KafkaMessage("body", key=b"1"), KafkaMessage("body", key=b"2"), "just a body message")

Probably, we can reuse the KafkaMessage class we have already. Also, please take a look at KafkaPublishCommand class

[ozeranskii]

Sorry, but magic {"message": ..., "key": ... } doesn't look as a good API. One reason - we may want to add more additional information to each method instead of just key (headers, publish timestamp, etc)

This reason we have a feature already - Response (or KafkaResponse)

@broker.subscriber(...)
@broker.publisher(...)
async def handler():
     return Response("body", key=b"key")

This object allows to setup any information of outgoing message you want

So, I suggest to support smth like this Response object - Message (we can discuss naming)

await broker.publish_batch(KafkaMessage("body", key=b"1"), KafkaMessage("body", key=b"2"), "just a body message")

Probably, we can reuse the KafkaMessage class we have already. Also, please take a look at KafkaPublishCommand class

I agree, I didn't like it either. I fixed it.