Consumer Group

Overview

A consumer group turns Events Store from a broadcast fan-out into a competing-consumers queue: subscribers sharing the same group split the stored events between them instead of each getting a copy of every event. Reach for this when a durable, ordered event log also needs to scale horizontally — a stream of order updates or audit records where one processor can't keep up, but each event still needs to be handled exactly once by the group as a whole.

It works by passing the same group: to subscribe_to_events_store on each subscriber alongside a start_at: option such as :start_new_only. The broker load-balances deliveries across every active member sharing that group and channel; adding another subscriber with the same group name is all it takes to add capacity. Gotchas: the start position belongs to the group's shared read cursor, not to any one subscriber — members joining later pick up wherever the group already is, not from the beginning. Different group names silently mean broadcast instead of load balancing, with no error to warn you. Delivery is exactly-once per group, but a crashed member's in-flight event isn't automatically handed to another member — design processing to be safely restartable.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-events-store.consumer-group"
group = "es-processing-group"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-es-group")

parent = self()

{:ok, sub1} =
  KubeMQ.Client.subscribe_to_events_store(client, channel,
    start_at: :start_new_only,
    group: group,
    on_event: fn event ->
      IO.puts("[Worker-1] #{event.body}")
      send(parent, {:worker, 1})
    end
  )

{:ok, sub2} =
  KubeMQ.Client.subscribe_to_events_store(client, channel,
    start_at: :start_new_only,
    group: group,
    on_event: fn event ->
      IO.puts("[Worker-2] #{event.body}")
      send(parent, {:worker, 2})
    end
  )

IO.puts("Two workers in group '#{group}' ready")
Process.sleep(500)

for i <- 1..6 do
  {:ok, _} = KubeMQ.Client.send_event_store(client,
    KubeMQ.EventStore.new(channel: channel, body: "Task #{i}"))
end

IO.puts("Sent 6 tasks. Waiting for distribution...")
Process.sleep(2_000)

KubeMQ.Subscription.cancel(sub1)
KubeMQ.Subscription.cancel(sub2)
KubeMQ.Client.close(client)

How It Works

• The group option assigns subscribers to a consumer group for load balancing
• Each event is delivered to exactly one subscriber within the group
• Both subscribers use the same start_at position for consistent replay behavior

Related

• Persistent Pub/Sub
• Events Consumer Group