[FEAT] Event Hubs: partition support ($management READ + partitioned consumer path)

[deyanp]

Service

Event Hubs (AMQP data plane + management plane)

API Action / Feature

Partition support: the $management READ operations (com.microsoft:eventhub, com.microsoft:partition) and the partitioned consumer path they feed.

Azure Documentation

• Event Hubs partitions
• AMQP 1.0 in Service Bus and Event Hubs — management-node request/response shape

Why is this needed?

Consuming from Event Hubs does not work at all today, for any SDK. Every consumer starts by asking the management node which partitions exist, and nothing answers:

get_eventhub_properties failed: Transport Implementation Error: StatusCode is nor found

With #237 (CBS + the Artemis library patches) publishing works end to end, so this is the last thing between floci-az and a working Event Hubs round-trip.

There is also a smaller correctness point: the entity config syntax accepts a partition count — floci.az.services.event-hub.entities is documented as "name:partitions", e.g. eh1:4 — and ArtemisConfigGenerator.parseEntities then discards it:

"The partition count in the entities string is accepted but ignored (Artemis handles routing)."

So a hub configured with four partitions is emulated as one undivided stream, silently.

Current state

grep -ri partition src/main/java/io/floci/az/services/eventhub/ returns exactly one hit — the comment above. There are no partition ids, no partition addresses, no partition properties. A hub becomes an Artemis address with one durable queue per consumer group.

The consumer path needs three things that do not exist yet:

1. Management replies. com.microsoft:eventhub READ must return name, created_at, partition_count and partition_ids; com.microsoft:partition READ must return begin_sequence_number, last_enqueued_sequence_number, last_enqueued_offset, last_enqueued_time_utc, is_partition_empty. A missing field fails the call outright.
2. Per-partition addresses. SDKs attach the receiver to amqps://{namespace}/{hub}/ConsumerGroups/{group}/Partitions/{id}. The generated topology has no /Partitions/ segment. Because Event Hubs sets auto-create-addresses=true, an attach today succeeds against an empty auto-created address and the consumer reads nothing — a silent wrong answer rather than an error.
3. Start positions. Earliest / latest / from-offset / from-sequence-number / from-enqueued-time arrive as AMQP selector filters over x-opt-* annotations, so those values have to exist on messages and be filterable.

Proposed implementation

Real N-partition emulation, using machinery that is already in the repo.

Topology — a durable queue per (consumer group, partition), named to match the address SDKs attach to:

<hub>/ConsumerGroups/<group>/Partitions/<n>

Routing and stamping — an Artemis broker plugin, the same mechanism as the Service Bus plugins already shipped in servicebus-artemis-extension.jar. On route it would:

1. choose the partition — an explicitly pinned partition id if present, else a stable hash of the partition key, else round-robin;
2. assign per-partition monotonic x-opt-sequence-number and x-opt-offset, and stamp x-opt-enqueued-time;
3. set the partition as an ordinary message property so each partition queue can select its own messages with a native Artemis filter.

Management responder — the $cbs pattern reused: a $management → $management-intercept divert plus a proton-j responder modelled on ServiceBusCbsResponder, answering both READ types from the counters the plugin maintains.

Start positions — once the annotations are real, filterable properties, Artemis selectors can serve all five forms.

Config — stop discarding the count in parseEntities and thread it through to topology and management.

This also lines the AMQP arm up with the Kafka arm, where Redpanda already has native partitions: a hub declared eh1:4 would mean the same thing over both protocols.

Design questions

Worth settling before any code, since these change the shape of it:

1. Is real multi-partition the direction you want? The cheaper option is to emulate a single partition and report partition_ids: ["0"], following the floci-az is always single-partition precedent in CosmosHandler. It would unblock consumers for far less work, but partitions are central to Event Hubs' consumer model — partition keys, per-partition ordering, per-partition checkpoints, EventProcessor distributing partitions across instances — so a single-partition emulator would make all of that pass against behaviour that cannot occur in production. My preference is real partitions for exactly that reason, but it is your call.
2. Validation of the configured count. It becomes load-bearing, and today it has no bounds at all — parseEntities splits on : and discards the number, so eh1:0, eh1:abc and eh1:10000 are silently equivalent. Reject zero and non-numeric (Azure requires at least one), and what upper bound? Each partition costs a durable queue per consumer group, so the count multiplies: eh1:10000 with five groups would be 50,000 queues. Azure caps at 32 per hub outside dedicated clusters, which seems a reasonable limit to borrow — cap, warn, or reject beyond it?
3. Broker plugin vs. an AMQP proxy for routing and stamping — the plugin looks right given the Service Bus precedent, but is there a reason you would avoid one here?
4. Persistence — in-memory per-namespace counters, consistent with storage: memory, or must sequence numbers and offsets survive a restart?
5. Partition assignment — does the hash need to match Azure's real assignment for a given partition key, or is any stable hash acceptable?

Are you willing to contribute a PR?

• Yes
• No

Happy to implement this once the direction above is settled — I would rather agree the semantics first than send a few hundred lines of Java that encodes the wrong model. Context: #227 reported the CBS gap, and #237 fixes CBS plus the max-message-size library patches, which is what made publishing work; this is the remaining half.

[deyanp]

Proposed answers to my own questions above, so there is something concrete to agree or push back on rather than five open ends. Happy to be overruled on any of them.

1. Real multi-partition. The deciding argument is cost, not purity: the expensive scaffolding is identical either way — broker plugin, management responder, topology generation, annotation stamping, selector filters. What N>1 adds on top is partition selection and a loop over queue names, perhaps 20–30% more work. So the choice is not "much cheaper vs much better", it is "slightly cheaper and semantically wrong vs slightly more work and correct". Given partitions are the whole Event Hubs consumer model — partition keys, per-partition ordering, per-partition checkpoints, EventProcessor distributing partitions across instances — the small saving is not worth making all of that pass against behaviour that cannot occur in Azure.

2. Validation only. With (1) answered this way, the "config says 4, emulator does 1" dilemma disappears: implement N, report N. What remains is bounding the count now that it is load-bearing — reject zero and non-numeric, and cap at Azure's own limit of 32 per hub outside dedicated clusters, since each partition costs a durable queue per consumer group and a typo multiplies.

3. Broker plugin, with the state kept out of it. The plugin is right for routing and stamping — the precedent exists and an AMQP proxy is the option rejected on #128 as heavy. The wrinkle is that the plugin runs inside the Artemis container while the responder runs in the floci JVM, so sharing sequence-number state across that boundary would be awkward. Suggestion: don't. Have the responder derive the management values by browsing over Jolokia, which floci already talks to:

Field	Derived from
is_partition_empty	queue message count
begin_sequence_number	first message's stamped annotation
last_enqueued_sequence_number / _offset / _time_utc	last message's annotations

4. No bespoke persistence. Falls out of (3): if the values are derived from queue contents, continuity follows Artemis's own journal and behaves correctly under memory and durable storage modes without special-casing either. The only stateful thing left is the next sequence number the plugin assigns, which it can seed from the last stamped message when a queue is opened.

5. Any stable hash — suggest Kafka's murmur2. Azure does not document its partition-key hash, and client code depends on the guarantee (same key → same partition, ordering within a partition), never on the index. A test asserting "key X lands in partition 2" would be asserting undocumented Azure internals. Using the same hash as the Redpanda arm would make eh1:4 distribute identically over both protocols, which turns the AMQP/Kafka alignment mentioned above into something real rather than merely structural.

One implementation note worth flagging early, found while verifying #237: because Event Hubs sets auto-create-addresses=true, a client attaching to an address the topology does not define succeeds against an empty auto-created address rather than failing. That makes addressing mistakes silent — a consumer reads nothing instead of erroring — so the generated topology needs to match the exact source addresses the SDKs use (amqps://{namespace}/{hub}/ConsumerGroups/{group}/Partitions/{id}), and it is probably worth turning auto-create off for these addresses so mismatches surface loudly.

I have started on an implementation along these lines and will open a PR; happy to rework it if you would rather a different shape.

[deyanp]

Implementation finding that changes the shape of this, so recording it before it gets lost.

I have partitions working end to end on the send side, and the $management READ answering — a consumer now gets partition_ids and attaches per partition. Verified at the broker: eight events over a four-partition hub with two consumer groups land two per partition in each group, one copy per group, nothing stranded.

But consumers still read nothing, and the reason is a grammar incompatibility rather than anything missing.

Every Event Hubs start position is sent as an AMQP selector over an annotation, quoted — including "earliest", which is literally amqp.annotation.x-opt-offset > '-1'. Two things in Artemis are in the way:

1. Annotation prefix. AMQPMessage.getObjectPropertyForFilter treats a filter key as an annotation only under the m. prefix (ANNOTATION_AREA_PREFIX = "m."). Azure's amqp.annotation. falls through to an ordinary property lookup. That part is workable — naming the properties exactly as the selectors reference them makes the lookup succeed, which is what my plugin now does.

2. The comparison itself, which is not workable from outside. ComparisonExpression.checkLessThanOperand rejects >, >=, <, <= against a string constant outright:

   if (value instanceof Number) return;
   // Else it's boolean or a String..
   throw new RuntimeException("Value '" + expr + "' cannot be compared.");

   Azure quotes its operands, so every start-position selector hits this. No property the broker stamps can satisfy it — the filter is rejected before any value is compared.

So supporting start positions needs the selector rewritten before Artemis parses it: amqp.annotation.x-opt-offset > '-1' → an unquoted numeric comparison against a property Artemis can read. That points at another artemis-amqp-patch in the AMQP protocol layer where the source filter is read, alongside the MSSBCBS one — I do not think it can be done from configuration or a broker plugin.

That is a bigger and more invasive piece than the rest, and it is squarely your call whether you want it, so I have stopped there rather than write it speculatively. Happy to take it on if you do.

Everything up to that point is working and I will open it as a draft PR so you can see the shape.

[deyanp]

Correcting my previous comment, and then closing this out: it is implemented and working. PR #242.

The diagnosis above was wrong about why. I said the blocker was ComparisonExpression.checkLessThanOperand rejecting > against a quoted constant, and that the annotation prefix was separately "workable" by naming properties as the selectors reference them. Neither holds. I ran every form through SelectorParser.parse against artemis-selector-2.44.0.jar:

REJECTED : amqp.annotation.x-opt-offset > '-1'        FilterException
REJECTED : amqp.annotation.x-opt-offset > -1          FilterException   <- unquoted, still rejected
REJECTED : m.x-opt-offset > -1                        FilterException   <- Artemis's own prefix
PARSED   : floci_partition = '0'

The unquoted form failing is what disproves it: nothing gets as far as comparing a value. The real blocker is the hyphens — x-opt-offset is not an identifier in the selector grammar, so it parses as x - opt - offset and the expression never parses at all. That also means naming a property amqp.annotation.x-opt-offset could never have worked, prefix or no prefix.

The conclusion survived the correction, and the fix is smaller than I expected. The selector still has to be rewritten before Artemis parses it, and it still belongs in the AMQP protocol layer — but it is one more <replace> in the artemis-amqp-patch execution that already exists for MSSBCBS, plus a small support class beside ServiceBusSessionSupport. No new mechanism, no new delivery path. The rewrite maps the three annotations onto underscored properties the plugin stamps, unquotes numeric operands, and turns @latest into a comparison against the clock read as the link attaches. The filter echoed back to the client is the one it sent.

Also worth recording, since the answers above got two things wrong that only showed up under a real SDK:

• Question 5 (partition-key hash). I suggested murmur2 for Kafka alignment. It would not have mattered: the key was never being read. The SDKs send it as a message annotation, getObjectProperty reads application properties only, and every keyed send was silently round-robining. Reading annotations first is the fix; the hash choice is String.hashCode(), whose stability the language spec pins.
• Question 3 (deriving management values over Jolokia). Not needed as it turned out — the responder answers com.microsoft:eventhub from configuration, and com.microsoft:partition with 501, which no SDK path this covers requires. Worth revisiting only if something turns out to need the per-partition runtime figures.

Round trip verified against the Azure SDKs for both Java and Rust: partition list discovered, events spread across four partitions, all read back per-partition from Earliest with offsets and sequence numbers intact. mocked: true has lost its reason, so PR #242 flips the shipped default — the JMS and Node compat suites that were silently skipping now run for real and pass.