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 name to subscribeToEventsStore on each subscriber alongside a start position from EventStoreStartPosition (such as StartFromNew or StartFromFirst). 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
• Node.js SDK installed (npm install kubemq-js)

Code

import { KubeMQClient, EventStoreStartPosition, createEventStoreMessage } from 'kubemq-js';

async function main() {
  const client = await KubeMQClient.create({
    address: 'localhost:50000',
    clientId: 'js-events-store-consumer-group-client',
  });

  try {
    const sub = client.subscribeToEventsStore({
      channel: 'js-events-store.consumer-group',
      group: 'workers',
      startFrom: EventStoreStartPosition.StartFromNew,
      onEvent: (event) => {
        console.log(`[group=workers] seq=${event.sequence}`, new TextDecoder().decode(event.body));
      },
      onError: (err) => {
        console.error('Error:', err.message);
      },
    });

    for (let i = 1; i <= 3; i++) {
      await client.sendEventStore(
        createEventStoreMessage({
          channel: 'js-events-store.consumer-group',
          body: `group-msg-${i}`,
        }),
      );
    }

    await new Promise((r) => setTimeout(r, 1000));
    sub.cancel();
  } finally {
    await client.close();
  }
}

main().catch(console.error);

How It Works

• group: 'workers' in subscribeToEventsStore() assigns this subscriber to the workers consumer group; KubeMQ load-balances events across all members.
• StartFromNew is used so no historical events are replayed — only events published after the subscription registers are delivered.
• In this single-subscriber example all 3 events go to the one sub; adding a second subscription with the same group name would cause KubeMQ to distribute the 3 events between both.
• The event.sequence in the log output is the server-assigned monotonic sequence number, which is shared across all group members for ordering guarantees.

Related

• Pattern overview
• Node.js SDK Reference
• Persistent Pub/Sub
• Cancel Subscription