dapr / dapr/components-contrib
RabbitMQ pub/sub: support the x-consistent-hash exchange type for partitioned per-key ordering
- Dominant language
- Go
- Stars
- 602
- Forks
- 580
- Avg merge
- 4d 9h
- Merged PRs (30d)
- 6
Description
## Describe the feature
Support the `x-consistent-hash` exchange type in the RabbitMQ pub/sub component, so a topic can be partitioned across consumers with per-key ordering preserved.
Previously requested in #3423, which was auto-closed as stale.
### Why the existing option isn't sufficient
The component's only in-the-box ordering control is `concurrencyMode: single`, which gives strict ordering via a *single serial consumer*. That is fine at modest throughput, but it does not scale out: ordering and parallelism become mutually exclusive.
The workload shape this matters for is a high-volume event stream — hundreds of millions of events a day and growing — where ordering is required *per key* (a tenant/device pair, say), not globally, and where processing must stay parallel. On Kafka that is partitioning. On RabbitMQ the equivalent is an `x-consistent-hash` exchange from the [`rabbitmq_consistent_hash_exchange`](https://github.com/rabbitmq/rabbitmq-server/tree/main/deps/rabbitmq_consistent_hash_exchange) plugin, which hashes a routing key to pick one bound queue, so every message for a key lands on the same queue and can be processed in order there while other keys proceed in parallel.
Today the component validates `exchangeKind` against the four built-in AMQP types only, so `x-consistent-hash` can be neither declared nor bound to.
### How it is used
Publish side is unchanged — the existing per-message `routingKey` metadata is the partition key, and the exchange hashes it:
```
client.PublishEvent(ctx, pubsub, topic, data,
dapr.PublishEventWithMetadata(map[string]string{"routingKey": "tenant-42/serial-7"}))
```
Subscribe side has one thing worth stating explicitly, because it is easy to get wrong: for a consistent-hash exchange the **binding key is the bound queue's bucket weight**, an integer — not a pattern. Each consumer binds its own queue with a weight, and the exchange distributes the key space across the bound queues in proportion:
```yaml
metadata:
routingKey: "10" # this queue's bucket weight on the hash ring
```
So the same `routingKey` metadata name carries a partition key when publishing and a bucket weight when subscribing. That asymmetry is inherent to the plugin, and deserves validation and documentation — an empty or non-numeric binding key currently fails deep inside the AMQP channel with a message that does not point at the configuration.
### Client support
No SDK work is needed. `amqp091-go` declares exchanges by an arbitrary kind string (`Channel.ExchangeDeclare(name, kind, …)`), so `x-consistent-hash` is accepted as-is; the plugin lives on the broker, not in the client. The component already carries `routingKey` metadata on both publish and `queue.bind`. What is missing is the validation allowlist and the binding-key semantics above.
### Scope
- Accept `x-consistent-hash` as an `exchangeKind`.
- Validate the subscription binding key as a positive integer bucket weight, and fail with a message naming the metadata field.
- Note the broker prerequisite in the docs: the `rabbitmq_consistent_hash_exchange` plugin must be enabled.
Related: #4553 — where the exchange is created by an external topology manager the component must also be able to bind to it rather than declare its own, and for `x-consistent-hash` that is the only workable path, since a declare of an unsupported kind can never match. The two are useful together.
## Release Note
RELEASE NOTE: **ADD** Support for the `x-consistent-hash` exchange type in the RabbitMQ pub/sub component, for partitioned per-key ordering with multiple consumers.
Contributor guide
Assessment
This issue has not been assessed yet.