dapr / dapr/components-contrib

RabbitMQ pub/sub: support the x-consistent-hash exchange type for partitioned per-key ordering

Open
#4,554 0 comments 0 reactions 1 assignee Claimed by @MyMirelHub View on GitHub
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.