Kafka PubSub: Propagate partition key to DLT

[denisbchrsk]

Description

Added support for partition key propagation to DLT when subscribing (including bulk subscribe) to a Kafka pubsub, by utilizing the enriched metadata from the subscribed message and use it to determine the partition key of the message produced to the DLT (utilizing the metadata which is added to the message in the doCallback and doBulkCallback according to this merged PR #3289).

Another change which was done in order to implement the partition key propagation when using bulk subscribe in Kafka, was passing each individual message's metadata to the DLT.
Previously, only the bulk request's metadata was considered when publishing to DLT, changing that allowed similar implementation added to Publish to be viable in BulkPublish as well.

Issue reference

We strive to have all PR being opened based on an issue, where the problem or feature have been discussed prior to implementation.

Please reference the issue this PR will close: #2227

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

• Code compiles correctly
• Created/updated tests
• Extended the documentation / Created issue in the https://github.com/dapr/docs/ repo: dapr/docs#[issue number]

[denisbchrsk]

Seems like the certification tests for RabbitMQ and MQTT3 are failing, I assume that this is a known issue?

[denisbchrsk]

Hey @yaron2 @berndverst,
This PR has been ready for quite a while now without being reviewed, is it possible to review this soon?

The issue with the partition key not being propagated to the DLT in Kafka pubsub is a blocker for us from using Dapr's Kafka pubsub, as DLTs are important for us to be able to handle errors separately.

[yaron2]

Hey @yaron2 @berndverst, This PR has been ready for quite a while now without being reviewed, is it possible to review this soon?

The issue with the partition key not being propagated to the DLT in Kafka pubsub is a blocker for us from using Dapr's Kafka pubsub, as DLTs are important for us to be able to handle errors separately.

This will be reviewed soon and if accepted will make it to the upcoming release

[yaron2]

Thanks for your contribution @denisbchrsk