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 setting the same group: on the EventsStoreSubscription passed to subscribe_to_events_store for each subscriber, alongside a start_position: such as EventStoreStartPosition::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
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-events-store.consumer-group'
group = 'es-workers'

begin
  client = KubeMQ::PubSubClient.new(address: address, client_id: 'es-group-example')
  puts "Connected to #{address}"

  cancel = KubeMQ::CancellationToken.new

  sub1 = KubeMQ::PubSub::EventsStoreSubscription.new(
    channel: channel,
    start_position: KubeMQ::PubSub::EventStoreStartPosition::START_NEW_ONLY,
    group: group
  )
  client.subscribe_to_events_store(sub1, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |event|
    puts "Worker-1 received: #{event.metadata}"
  end

  sub2 = KubeMQ::PubSub::EventsStoreSubscription.new(
    channel: channel,
    start_position: KubeMQ::PubSub::EventStoreStartPosition::START_NEW_ONLY,
    group: group
  )
  client.subscribe_to_events_store(sub2, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |event|
    puts "Worker-2 received: #{event.metadata}"
  end
  sleep 1

  5.times do |i|
    msg = KubeMQ::PubSub::EventStoreMessage.new(channel: channel, metadata: "es event #{i}", body: "data-#{i}")
    client.send_event_store(msg)
    puts "Sent event #{i}"
  end

  sleep 2
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  cancel&.cancel
  client&.close
  puts 'Done'
end

How It Works

• The group parameter enables load-balanced delivery — each event goes to exactly one group member.
• Both workers share the same start position and group name.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Persistent Pub/Sub