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. 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.

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

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).
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.

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). 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).

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.