Consumer-Group Event Broker
Problem
Several worker services consume the same event stream independently. Within one service, its workers form a consumer group and divide the topic's partitions so that only one member handles a partition at a time. A record remains available for other groups even after one group acknowledges it.
Design EventBroker, a bounded in-memory broker that manages these partitioned topics, consumer groups, acknowledgements, retries, and rebalancing. One broker instance is shared by multiple threads.
Requirements
- Bounded partition logs. Each partition retains at most the configured capacity. A successful publication appends to its specified partition with the next offset; a full partition returns
-1without consuming an offset. - Deterministic ownership. A non-empty group assigns every partition to exactly one member by sorting member identifiers in ASCII order and assigning partition
pto memberp mod memberCount. Repeated joins and absent leaves do not change membership. - Ordered exclusive delivery. A member polls only its assigned partitions, receives at most one in-flight record per partition, and cannot receive a later partition offset before acknowledging the current one. Poll results use ascending partition order.
- Acknowledge or retry. Only the current owner may acknowledge or retry the exact in-flight record. Acknowledgement advances committed progress by one; retry clears the delivery without advancing it. Any mismatch returns
falsewithout changing state. - Progress-preserving rebalance. Membership changes replace the complete assignment atomically. A moved partition loses its in-flight delivery but keeps its committed offset. A group and its progress remain after its last member leaves.
- Shared retention. A record is removed only after every existing group for its topic has acknowledged it. Empty groups still count. With no groups, records remain retained. A new group starts before each partition's earliest retained offset, or before its next offset if that partition is empty.
- Atomic shared use. Each public call takes effect atomically at one point between its invocation and return, and calls respect program order within each thread. This guarantee is called linearizability. A call does not wait for a future record or for membership, capacity, or other broker state to change, but it may wait to acquire internal synchronization that protects the current state. Concurrent calls preserve all broker invariants.
API
BrokerRecord is a value snapshot with fields topic: string, partition: int32, offset: int64, and payload: string. Mutating a returned value where the language permits it must not change broker state.
| Signature | Returns | Behavior |
|---|---|---|
EventBroker(capacityPerPartition: int32) | Not applicable | Creates an empty broker. Throws the runtime's invalid-argument exception unless capacity is from 1 through 1,000. |
create(topic: string, partitionCount: int32) | boolean | Creates a topic with empty fixed partitions. Returns false and preserves the existing topic on a duplicate. |
join(topic: string, group: string, member: string) | int32[] | Creates the group if absent, adds the member if needed, rebalances, and returns that member's current partitions in ascending order. |
leave(topic: string, group: string, member: string) | boolean | Removes an existing member and rebalances. Returns false when the group or member is absent. |
publish(topic: string, partition: int32, payload: string) | int64 | Appends the payload and returns its partition-local offset, or returns -1 when the partition is full. Empty payloads are valid. |
poll(topic: string, group: string, member: string, maxRecords: int32) | BrokerRecord[] | Reserves up to the limit from the member's eligible partitions. Returns an empty list for an absent group or member, no eligible record, or assignments already in flight. |
ack(topic: string, group: string, member: string, partition: int32, offset: int64) | boolean | Commits and clears the exact matching delivery, then removes any newly eligible shared retained prefix. Returns false on a state mismatch. |
retry(topic: string, group: string, member: string, partition: int32, offset: int64) | boolean | Clears the exact matching delivery without committing it. Returns false on a state mismatch. |
Identifiers must contain 1 to 64 ASCII letters, digits, _, or -. They are case-sensitive and are not trimmed. Every method other than create throws the runtime's invalid-argument exception for an unknown topic. Methods also throw that exception for a null string where supported, an invalid identifier, an invalid partition, a negative acknowledgement or retry offset, or an out-of-range maxRecords or partitionCount. Invalid calls have no side effects. The exceptions are Python ValueError, Java IllegalArgumentException, and C++ std::invalid_argument.
Examples
With capacity 2:
| Step | Operation | Result |
|---|---|---|
| 1 | create("orders", 1) | true |
| 2 | publish("orders", 0, "a") | 0 |
| 3 | publish("orders", 0, "b") | 1 |
| 4 | join("orders", "billing", "m1") | [0] |
| 5 | poll("orders", "billing", "m1", 2) | [BrokerRecord("orders", 0, 0, "a")] |
| 6 | retry("orders", "billing", "m1", 0, 0) | true |
| 7 | poll("orders", "billing", "m1", 2) | [BrokerRecord("orders", 0, 0, "a")] |
| 8 | ack("orders", "billing", "m1", 0, 0) | true |
| 9 | poll("orders", "billing", "m1", 2) | [BrokerRecord("orders", 0, 1, "b")] |
For a three-partition topic, member a owns [0, 1, 2]. After b joins, a owns [0, 2] and b owns [1]. An unacknowledged delivery from partition 1 to a is canceled; b can poll that offset, and a later acknowledgement from a returns false.
Constraints
1 <= capacityPerPartition <= 1,0001 <= partitionCount <= 641 <= maxRecords <= 64- At most 32 topics, 128 groups per topic, 64 distinct members per group, and 50,000 calls per testcase.
- Payloads are non-null strings of at most 1,024 UTF-8 bytes. An empty payload is valid.
- Offsets have signed 64-bit semantics, remain nonnegative, and cannot overflow within the call limit.
Notes
Concurrent publications to one partition receive distinct consecutive offsets, limited by available capacity. Concurrent polls cannot reserve the same group-partition delivery twice, while different groups may each reserve the same record. Rebalance is atomic with polling, acknowledgement, and retry. Acknowledgement and retry for one delivery cannot both succeed. Capacity cleanup is atomic with publication. No fairness order is promised, and implementations must not deadlock.