--- title: "Consume events" slug: "dev-event-broker-guides-consuming" description: "Learn to read VAST Event Broker topics with Kafka consumers, covering groups, offsets, rebalancing, and more using the Confluent Kafka Python client." updated: 2026-10-30T18:18:46Z published: 2026-10-30T18:18:46Z canonical: "docs.vastdata.com/dev-event-broker-guides-consuming" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Consume events *Applies to VAST Cluster 5.5.* This guide shows how to read VAST Event Broker topics with a Kafka consumer: consumer groups, committing offsets, rebalancing, reading a partition directly, seeking, what isolation level you get, and deleting a group. The examples use the Confluent Kafka Python client (`confluent-kafka`). Run the code blocks in order in one Python session. Below each block is the output you should see. ## Set up You need the environment variables from the [quickstart](/develop/docs/dev-event-broker-quickstart#before-you-start). If authorization is turned on for the view, a consumer group needs `KafkaRead` on the topic and on the group, and deleting a group needs `KafkaDelete` on the group; see [security](/develop/docs/dev-event-broker-operations-security). This block creates a topic with three partitions, produces 30 events to it, and defines a helper that polls a bounded number of times: ``` import json import os from confluent_kafka import OFFSET_BEGINNING, Consumer, KafkaException, Producer, TopicPartition from confluent_kafka.admin import AdminClient, NewTopic TOPIC = "docs-consuming-orders" GROUP = "docs-consuming-billing" GROUP_2 = "docs-consuming-rebalance" def kafka_conf(extra=None): conf = {"bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"]} if os.environ.get("KAFKA_USERNAME"): conf.update({ "security.protocol": "SASL_PLAINTEXT", "sasl.mechanism": "PLAIN", "sasl.username": os.environ["KAFKA_USERNAME"], "sasl.password": os.environ["KAFKA_PASSWORD"], }) conf.update(extra or {}) return conf def consumer_for(group): return Consumer(kafka_conf({ "group.id": group, "auto.offset.reset": "earliest", "enable.auto.commit": False, })) def read(consumer, expected, max_polls=60): """Poll until `expected` events have arrived or `max_polls` polls have passed.""" events = [] for _ in range(max_polls): msg = consumer.poll(1.0) if msg is None: if len(events) >= expected: break continue if msg.error(): raise KafkaException(msg.error()) events.append((msg.partition(), msg.offset(), json.loads(msg.value()))) return events admin = AdminClient(kafka_conf()) admin.create_topics([NewTopic(TOPIC, num_partitions=3, replication_factor=1)], operation_timeout=30)[TOPIC].result() producer = Producer(kafka_conf({"acks": "all", "enable.idempotence": False, "compression.type": "none"})) def produce(first, count): for n in range(first, first + count): producer.produce(TOPIC, key=f"order-{n}", value=json.dumps({"order": n})) return producer.flush(30) print("still queued:", produce(0, 30)) ``` ``` still queued: 0 ``` ## Consume in a group ![A topic with four partitions is read by a consumer group with two consumers: consumer 1 has partitions 0 and 1, consumer 2 has partitions 2 and 3. For each partition the cluster stores the group's committed offset, the next offset the group will read. The lag is the end offset minus the committed offset. When a consumer joins or leaves, the group rebalances eagerly: every partition is revoked and then assigned again among the members.](https://cdn.us.document360.io/89b91113-5dc6-49db-86ec-57b16dc8ce56/Images/Documentation/eb-consumer-group.svg?sv=2026-02-06&spr=https&st=2026-11-08T08%3A55%3A31Z&se=2026-11-08T09%3A55%3A31Z&sr=c&sp=r&sig=%2BUPIvmvCe%2FVIFyTPqFZ5ZlPOphk3e1vx%2FwyauoguleE%3D) *Each partition goes to one consumer in the group, and the cluster keeps the group's committed offsets.* Subscribe, process the events, then commit the offsets: ``` consumer = consumer_for(GROUP) consumer.subscribe([TOPIC]) events = read(consumer, expected=30) consumer.commit(asynchronous=False) print("events:", len(events)) print("orders:", sorted(e["order"] for _, _, e in events) == list(range(30))) print("partitions:", sorted({p for p, _, _ in events})) ``` ``` events: 30 orders: True partitions: [0, 1, 2] ``` The first poll waits until the group has formed, which takes a few seconds. Events are in order within each partition; there's no order across partitions. Commit after you've processed the events, not before, so that a crash makes you read them again rather than skip them. This page turns off auto-commit and commits explicitly. If you keep `enable.auto.commit=true` (the default), the client commits in the background on a timer. ### Check the committed offsets The committed offset is the next offset the group will read, per partition: ``` committed = consumer.committed([TopicPartition(TOPIC, p) for p in range(3)], timeout=10) for tp in committed: low, high = consumer.get_watermark_offsets(tp, timeout=10) print(f"partition {tp.partition}: committed={tp.offset} end={high} lag={high - tp.offset}") consumer.close() ``` ``` partition 0: committed=13 end=13 lag=0 partition 1: committed=6 end=6 lag=0 partition 2: committed=11 end=11 lag=0 ``` `get_watermark_offsets()` returns the earliest offset and the end offset (the offset the next event will get). The difference between the end offset and the committed offset is the group's lag. Offsets are per partition and increase, but they can have gaps, for example after compaction, so don't count events by subtracting offsets. The end offset that the broker reports (through `get_watermark_offsets()`, or a `ListOffsets` request for the latest offset) can trail the newest events by tens of seconds: it can catch up 10 to 55 seconds after a burst of writes, while a consumer can already read the new events. So lag computed right after producing can read too low, even zero. Don't use the end offset to decide whether a consumer has caught up; read until a poll returns nothing. ### Resume where the group left off A new consumer in the same group starts from the committed offsets. It gets only the events produced since the commit: ``` produce(30, 5) consumer = consumer_for(GROUP) consumer.subscribe([TOPIC]) events = read(consumer, expected=5) consumer.commit(asynchronous=False) consumer.close() print("orders:", sorted(e["order"] for _, _, e in events)) ``` ``` orders: [30, 31, 32, 33, 34] ``` `auto.offset.reset` decides where a group starts when it has no committed offset for a partition, or when its committed offset is no longer in the topic because retention removed those events: `earliest` or `latest`. Committed offsets don't expire with time; they stay until you delete the group or the topic. ## Rebalancing When a consumer joins or leaves a group, the group rebalances: the broker stops the group, and the partitions are shared out again among the members. VAST Event Broker supports the classic consumer group protocol, with eager rebalancing: - Use an eager assignor: `range` or `roundrobin` (the Confluent client's default is `range,roundrobin`). Cooperative rebalancing (`cooperative-sticky`) isn't supported ([KB](/docs-5-5/docs/publishing-events-to-vast-event-broker)). - `group.instance.id` is accepted but ignored: static membership has no effect, and the consumer joins as an ordinary member. - Don't set `group.protocol=consumer`: the newer consumer group protocol isn't supported. Leave it at `classic`, the default in the Confluent client 2.4 to 2.8. - A group has at most as many active consumers as the topic has partitions. Extra consumers get no partitions. Two consumers in one group share the three partitions. In one Python session both have to keep polling while the group rebalances: ``` c1, c2 = consumer_for(GROUP_2), consumer_for(GROUP_2) c1.subscribe([TOPIC]) c2.subscribe([TOPIC]) for _ in range(60): # bounded: poll both until the partitions are shared out c1.poll(0.5) c2.poll(0.5) sizes = sorted([len(c1.assignment()), len(c2.assignment())]) if sizes[0] > 0 and sum(sizes) == 3: break print("partitions per consumer:", sizes) ``` ``` partitions per consumer: [1, 2] ``` To commit what you've processed before your partitions move to another consumer, pass an `on_revoke` callback to `subscribe()` and commit in it. Each consumer must call `poll()` more often than `max.poll.interval.ms`, or the group removes it and rebalances. ## Read partitions without a group With `assign()` instead of `subscribe()`, a consumer reads the partitions you name, from the offsets you give, without joining a group. Use it for replays, tools and tests. This reads the fullest partition from its earliest offset: ``` reader = consumer_for("docs-consuming-replay") sizes = {} for p in range(3): low, high = reader.get_watermark_offsets(TopicPartition(TOPIC, p), timeout=10) sizes[p] = (low, high) part = max(sizes, key=lambda p: sizes[p][1] - sizes[p][0]) low, high = sizes[part] reader.assign([TopicPartition(TOPIC, part, OFFSET_BEGINNING)]) events = read(reader, expected=high - low) print("starts at the earliest offset:", events[0][1] == low, "| all read:", len(events) == high - low) ``` ``` starts at the earliest offset: True | all read: True ``` The broker keeps no group for an assigned consumer unless it commits offsets; a commit creates the group. ### Seek to an offset `seek()` moves an assigned partition to another offset. The next poll reads from there: ``` target = events[2][1] # the third event's offset in that partition reader.seek(TopicPartition(TOPIC, part, target)) msg = reader.poll(10.0) print("offset after seek:", msg.offset() == target) reader.close() ``` ``` offset after seek: True ``` You can seek to an offset, to the beginning (`OFFSET_BEGINNING`), to the end (`OFFSET_END`) or by time. `offsets_for_times()` returns, for each partition, the first offset whose timestamp is at or after the time you ask for, and `-1` for a time after the newest event. To select events by time range, you can also query the topic as a table and filter on its timestamp columns; see [query topics](/develop/docs/dev-event-broker-guides-querying-topics). ## Isolation and fetching - **Consumers see committed events only.** A producer's request is written as one transaction, and a consumer never reads part of one. There are no Kafka transactions, so `isolation.level=read_uncommitted` isn't supported; the broker accepts it but still returns only committed events. - **Every fetch is a full fetch.** Incremental fetch sessions aren't supported. Clients detect this and send full fetch requests, with no change to your code. - **Delayed fetch settings have no effect.** When there's nothing new, the broker answers straight away instead of waiting `fetch.wait.max.ms`, and it ignores `fetch.min.bytes`. A consumer on a quiet topic therefore sends more fetch requests; that's expected. - **No rack awareness.** `client.rack` has no effect: each partition is read through one broker address. Consuming isn't subject to VAST protocol auditing or QoS. ## Delete a consumer group You can delete a group once it has no active members. Deleting it removes its committed offsets. With the two consumers still in the group, the delete fails: ``` def delete_group(group): future = admin.delete_consumer_groups([group], request_timeout=30)[group] try: future.result() return "deleted" except KafkaException as e: return e.args[0].name() print("while active:", delete_group(GROUP_2)) c1.close() c2.close() print("after close: ", delete_group(GROUP_2)) ``` ``` while active: NON_EMPTY_GROUP after close: deleted ``` Things to know about groups: - A Kafka-enabled view supports up to 256 consumer groups, and listing groups returns at most 256 ([KB](/docs-5-5/docs/publishing-events-to-vast-event-broker)). Delete groups you no longer need, such as ones from tests. - Deleting individual committed offsets (`OffsetDelete`) isn't supported; delete the group instead. - Committed offsets aren't replicated. After a failover to a replication peer, groups start again from their `auto.offset.reset` setting ([KB](/docs-5-5/docs/publishing-events-to-vast-event-broker)). ## Clean up ``` print("billing group:", delete_group(GROUP)) for name, future in admin.delete_topics([TOPIC], operation_timeout=30).items(): future.result() print("deleted", name) ``` ``` billing group: deleted deleted docs-consuming-orders ``` Deleting a topic also deletes every group's committed offsets for it. ## See also - [Producing](/develop/docs/dev-event-broker-guides-producing): keys, partitions and delivery guarantees. - [Query topics](/develop/docs/dev-event-broker-guides-querying-topics): read a topic as a table, filter by time. - [Concepts](/develop/docs/dev-event-broker-concepts-index): partitions, offsets and consumer groups. - [Monitoring](/develop/docs/dev-event-broker-operations-monitoring): check consumer lag. - [Kafka compatibility](/develop/docs/dev-event-broker-reference-kafka-compatibility) and [limits](/develop/docs/dev-event-broker-reference-limits).