--- title: "Quickstart" slug: "dev-event-broker-quickstart" description: "Create a Kafka topic, produce JSON events, and consume them using the Confluent Kafka Python client and VAST DataBase SDK in just 10 minutes." updated: 2026-10-30T18:19:06Z published: 2026-10-30T18:19:06Z canonical: "docs.vastdata.com/dev-event-broker-quickstart" --- > ## 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. # Quickstart *Applies to VAST Cluster 5.5.* In this quickstart you'll create a topic with a Kafka admin client, produce a few JSON events, and consume them in a consumer group. Then you'll read the same topic as a VAST DataBase table with the Python SDK, and remove what you created. You use the Confluent Kafka Python client (`confluent-kafka`) and the VAST DataBase SDK (`vastdb`). There's one path and no optional settings. It takes about 10 minutes once your platform team has set up access. Run each code block in order in the same Python session. Below each block is the output you should see. ## Before you start You need: - A Kafka-enabled view (the broker), with a virtual IP pool, set up by your platform team. Ask them for three of the pool's virtual IPs as bootstrap servers. Port 9092 takes unencrypted connections and 9093 takes TLS. See [setup](/develop/docs/dev-event-broker-operations-setup) for the full list of what to ask for. - If the view requires SASL/PLAIN, a VAST local user name and password. This page uses an unencrypted connection; for TLS and mTLS see [security](/develop/docs/dev-event-broker-operations-security). Your platform team sets up the broker's TLS certificate and the identity policies that grant Kafka permissions ([setup](/develop/docs/dev-event-broker-operations-setup)). - The name of the view's S3 bucket, and S3 access keys with read access to it, for the SDK part. See VAST DataBase [setup prerequisites](/develop/docs/dev-database-operations-setup-prerequisites). - Python 3.10 or later. Install the clients. VAST Cluster 5.5 supports Confluent Kafka Python client 2.4 to 2.8 ([KB](/docs-5-5/docs/publishing-events-to-vast-event-broker)): ``` pip install "confluent-kafka>=2.4,<2.9" vastdb ``` Set the connection settings in your environment: ``` export KAFKA_BOOTSTRAP=":9092,:9092,:9092" # Only if the view requires SASL/PLAIN: export KAFKA_USERNAME="" export KAFKA_PASSWORD="" # For the VAST DataBase SDK: export AWS_ACCESS_KEY_ID="" export AWS_SECRET_ACCESS_KEY="" export AWS_S3_ENDPOINT_URL="http://" export EVENTS_BUCKET="" ``` ![Six steps, in order, then your application connects. 1, platform team: a VIP pool with the Protocols role and at least one virtual IP per CNode, created by a cluster administrator; a Kafka view has exactly one, which can be modified but not replaced. 2, platform team: the Kafka view on a new path, with its bucket owner user and the VIP pool; enabling Kafka also enables S3 and DATABASE, and the bucket name is the name of the database that holds your topics. 3, platform team: connections and authentication: unencrypted on port 9092, TLS on port 9093, or both; no authentication, SASL/PLAIN with a VAST local user, or mTLS; and the CA certificate for TLS. 4, platform team: if authorization is on, an identity policy with the Kafka actions you need, such as KafkaRead and KafkaWrite on topics and KafkaRead on consumer groups. 5, platform team or you: topics, with partition counts, which are fixed when a topic is created, and retention, or the right to create topics with the Kafka Admin API. 6, platform team: for reading topics as tables and for blob expansion, an S3 access key pair, the database endpoint, read access to the view's bucket and the s3:Tabular*BlobExpansion actions. Then your application connects: bootstrap.servers is three virtual IPs from the pool, as vip:port, every virtual IP in the pool must be reachable, and producers set compression.type to none and enable.idempotence to false.](https://cdn.us.document360.io/89b91113-5dc6-49db-86ec-57b16dc8ce56/Images/Documentation/eb-setup-flow.svg?sv=2026-02-06&spr=https&st=2026-11-08T08%3A55%3A09Z&se=2026-11-08T09%3A55%3A09Z&sr=c&sp=r&sig=8A0%2BIk47AoXAr%2BI8U7gisoXIuJ%2BQZitoQNrRHhic%2FaQ%3D) *What your platform team sets up for a broker, in order, before your Kafka client connects.* See [setup](/develop/docs/dev-event-broker-operations-setup) for each item. ## Set up Load the settings, and check that the client can reach the broker: ``` import json import os import vastdb from confluent_kafka import Consumer, KafkaException, Producer from confluent_kafka.admin import AdminClient, NewTopic TOPIC = "docs-quickstart-events" GROUP = "docs-quickstart-group" BUCKET = os.environ["EVENTS_BUCKET"] 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 admin = AdminClient(kafka_conf()) metadata = admin.list_topics(timeout=10) print("connected:", len(metadata.brokers) > 0) ``` ``` connected: True ``` Every virtual IP in the view's pool is a broker, so the metadata lists one broker per virtual IP. ## Create a topic Topics aren't created automatically, so create one before you produce to it. This one has two partitions: ``` futures = admin.create_topics( [NewTopic(TOPIC, num_partitions=2, replication_factor=1)], operation_timeout=30, ) for name, future in futures.items(): future.result() print("created", name) partitions = admin.list_topics(TOPIC, timeout=10).topics[TOPIC].partitions print("partitions:", sorted(partitions)) ``` ``` created docs-quickstart-events partitions: [0, 1] ``` The partition count is fixed when you create the topic. See [topics](/develop/docs/dev-event-broker-guides-topics). ## Produce events Produce six JSON events with a key each. The producer turns off idempotence and compression, which VAST Event Broker doesn't support: ``` producer = Producer(kafka_conf({ "acks": "all", "enable.idempotence": False, "compression.type": "none", })) delivered, failed = [], [] def on_delivery(err, msg): if err is not None: failed.append(err.name()) else: delivered.append(msg.key().decode()) events = [ {"id": 1, "user": "ana", "action": "login"}, {"id": 2, "user": "ben", "action": "view"}, {"id": 3, "user": "ana", "action": "purchase"}, {"id": 4, "user": "chen", "action": "login"}, {"id": 5, "user": "ben", "action": "logout"}, {"id": 6, "user": "chen", "action": "view"}, ] for event in events: producer.produce(TOPIC, key=event["user"], value=json.dumps(event), on_delivery=on_delivery) remaining = producer.flush(30) print("delivered:", len(delivered), "failed:", failed, "still queued:", remaining) ``` ``` delivered: 6 failed: [] still queued: 0 ``` With `acks=all`, a delivery report means the events are committed to the topic. Events with the same key go to the same partition. See [producing](/develop/docs/dev-event-broker-guides-producing). ## Consume events in a group Read the events back in a consumer group, from the start of the topic, and commit the offsets: ``` consumer = Consumer(kafka_conf({ "group.id": GROUP, "auto.offset.reset": "earliest", "enable.auto.commit": False, })) consumer.subscribe([TOPIC]) received = [] for _ in range(60): # poll for up to about a minute msg = consumer.poll(1.0) if msg is None: if len(received) == len(events): break continue if msg.error(): raise KafkaException(msg.error()) received.append(json.loads(msg.value())) consumer.commit(asynchronous=False) consumer.close() for event in sorted(received, key=lambda e: e["id"]): print(event) ``` ``` {'id': 1, 'user': 'ana', 'action': 'login'} {'id': 2, 'user': 'ben', 'action': 'view'} {'id': 3, 'user': 'ana', 'action': 'purchase'} {'id': 4, 'user': 'chen', 'action': 'login'} {'id': 5, 'user': 'ben', 'action': 'logout'} {'id': 6, 'user': 'chen', 'action': 'view'} ``` The first poll in a new group waits until the group has formed, so allow a few seconds before the events arrive. Order is kept within a partition, not across partitions; the output above is sorted by `id`. See [consuming](/develop/docs/dev-event-broker-guides-consuming). ## Read the topic as a table A topic is a VAST DataBase table, in the schema `kafka_topics` of the Kafka view's bucket. The table has the topic's name. Open it with the SDK and look at its columns: ``` session = vastdb.connect() with session.transaction() as tx: table = tx.bucket(BUCKET).schema("kafka_topics").table(TOPIC) for field in table.columns(): print(field.name, field.type) ``` ``` key binary value binary timestamp_broker timestamp[ms] timestamp_producer timestamp[ms] headers list not null> external_offset int64 compaction_process_seq_num int64 extra_metadata binary ``` Each row is one event. `key` and `value` hold the bytes the producer sent, and there are columns for the broker and producer timestamps and the headers. The last three columns are used by the cluster itself, and are empty for events from Kafka producers. See [querying topics](/develop/docs/dev-event-broker-guides-querying-topics#columns). Read the key and value columns, and decode the JSON yourself: ``` with session.transaction() as tx: table = tx.bucket(BUCKET).schema("kafka_topics").table(TOPIC) rows = table.select(columns=["key", "value"]).read_all().to_pylist() for row in sorted(rows, key=lambda r: json.loads(r["value"])["id"]): print(row["key"].decode(), json.loads(row["value"])) ``` ``` ana {'id': 1, 'user': 'ana', 'action': 'login'} ben {'id': 2, 'user': 'ben', 'action': 'view'} ana {'id': 3, 'user': 'ana', 'action': 'purchase'} chen {'id': 4, 'user': 'chen', 'action': 'login'} ben {'id': 5, 'user': 'ben', 'action': 'logout'} chen {'id': 6, 'user': 'chen', 'action': 'view'} ``` The table shows committed events only, the same ones a consumer can read. It's read-only through the VAST DataBase API: you add events by producing them. To query topics with SQL through the VAST Query Engine or Trino, or to turn JSON values into typed columns as they arrive, see [query topics](/develop/docs/dev-event-broker-guides-querying-topics) and [blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion). ## Clean up Delete the consumer group and the topic: ``` for name, future in admin.delete_consumer_groups([GROUP], request_timeout=30).items(): future.result() print("deleted group", name) for name, future in admin.delete_topics([TOPIC], operation_timeout=30).items(): future.result() print("deleted topic", name) ``` ``` deleted group docs-quickstart-group deleted topic docs-quickstart-events ``` Deleting a topic deletes its table and all its events. ## See also - [Concepts](/develop/docs/dev-event-broker-concepts-index): views, topics, partitions, offsets and consumer groups. - [Topics](/develop/docs/dev-event-broker-guides-topics), [producing](/develop/docs/dev-event-broker-guides-producing) and [consuming](/develop/docs/dev-event-broker-guides-consuming). - [Query topics](/develop/docs/dev-event-broker-guides-querying-topics) with the SDK, the Query Engine and Trino. - [Kafka compatibility](/develop/docs/dev-event-broker-reference-kafka-compatibility): what's supported, and what a client sees when it isn't.