--- title: "CI/CD and automation" slug: "dev-event-broker-operations-ci-cd" description: "Manage VAST Event Broker objects with Kafka views, topics, and blob expansions using VMS REST API, vastpy, and Terraform for efficient data handling." updated: 2026-10-30T18:19:06Z published: 2026-10-30T18:19:06Z canonical: "docs.vastdata.com/dev-event-broker-operations-ci-cd" --- > ## 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. # CI/CD and automation *Applies to VAST Cluster 5.5.* This page shows how to manage VAST Event Broker objects from a pipeline: the Kafka view, its topics, and blob expansions that turn a topic's JSON events into table columns. It covers the VMS REST API, `vastpy`, the VAST Terraform provider, and what the VAST admin MCP server can and can't do. Every example was run on a VAST Cluster 5.5.0-SP6 lab cluster. Application code can also create and configure topics with the Kafka Admin API; see [manage topics](/develop/docs/dev-event-broker-guides-topics). The tools on this page are the right choice when topics are infrastructure: declared in Git, reviewed, and applied by a pipeline. For the shared setup (API tokens, TLS, users, keys and Terraform state), see [CI/CD and automation for VAST DataBase](/develop/docs/dev-database-operations-ci-cd). To choose between the tools, see [automation tools compared](/develop/docs/dev-database-reference-automation-tools). ## What each tool can do | Task | REST API | `vastpy` | Terraform (4.0.2) | VAST admin MCP server | | --- | --- | --- | --- | --- | | Create a Kafka view | Yes, on a VIP pool from your platform team | Yes | `vastdata_view` | No: the create fails | | Create, change and delete topics | Yes | Yes | `vastdata_topic` | No tools | | Compacted topics, `max_message_bytes` | Yes | Yes | No arguments for them | No | | Create a blob expansion | Yes, after its target table exists | Yes | `vastdata_blob_expansion`, after its target table exists | No tools | | Create the target table | Yes (`/schemas/`, `/tables/`) | Yes | No resource | No | | List Kafka views | Yes | Yes | Data source | `list_views_vast` | `vastdata_kafka_broker` in the Terraform provider registers an **external** (third-party) event broker with the cluster. It doesn't create or configure VAST Event Broker; use `vastdata_view` and `vastdata_topic` for that. ## Before you start **Admin task: send to your platform team.** Two things on this page need them: - **A VIP pool for each Kafka view.** A cluster administrator creates it, with the Protocols role. See [Configuring VAST Event Broker](/docs-5-5/docs/publishing-events-to-vast-event-broker#configuring-vast-event-broker) in the KB and [setup](/develop/docs/dev-event-broker-operations-setup). A pool that another Kafka view already uses only accepts views that require mTLS, so a second view with unencrypted connections needs its own pool ([below](/develop/docs/dev-event-broker-operations-ci-cd#create-the-kafka-view)). - **An API token for the automation identity.** The examples use a tenant administrator's VMS API token. See [Managing VMS API tokens](/docs-5-5/docs/managing-vms-api-tokens) and [CI/CD and automation for VAST DataBase](/develop/docs/dev-database-operations-ci-cd). Call a versioned API path, `/api/v8/` or `/api/latest/` (they return the same objects on 5.5). The unversioned `/api/views/` returns an older shape of the view without any of the `kafka_*` fields, so a script that reads or copies a view through it drops the Event Broker settings without an error. `vastpy` uses `/api/latest/`. ### API documentation - The KB's [VAST Cluster 5.5 REST API reference](https://kb.vastdata.com/api-5-5/apidocs): [Create View](https://kb.vastdata.com/api-5-5/apidocs/create-view), [List Kafka Topics](https://kb.vastdata.com/api-5-5/apidocs/list-kafka-topics), [Create a Kafka Topic](https://kb.vastdata.com/api-5-5/apidocs/create-a-kafka-topic), [Modify a Kafka Topic](https://kb.vastdata.com/api-5-5/apidocs/modify-a-kafka-topic), [Show a Kafka Topic](https://kb.vastdata.com/api-5-5/apidocs/show-a-kafka-topic), [Delete a Kafka Topic](https://kb.vastdata.com/api-5-5/apidocs/delete-a-kafka-topic), [Create a Blob Expansion](https://kb.vastdata.com/api-5-5/apidocs/create-a-blob-expansion), [Show a Blob Expansion](https://kb.vastdata.com/api-5-5/apidocs/show-a-blob-expansion) and [Add Columns to a Blob Expansion](https://kb.vastdata.com/api-5-5/apidocs/add-columns-to-a-blob-expansion). - Your cluster's own copy at `https://<VMS address>/docs/` (VMS REST API documentation). The Swagger UI at `https://<VMS address>/api/` lists only a few endpoints, not these. ## REST API The examples use `VMS`, the VMS address, and `VMS_TOKEN`, a tenant admin's API token. `-k` skips TLS verification for a lab cluster with a self-signed certificate; in a pipeline, pass the cluster's CA with `--cacert` instead. ### Create the Kafka view A Kafka view has the `KAFKA`, `S3` and `DATABASE` protocols, a bucket owner, an S3 Native view policy, and its VIP pool in `kafka_vip_pools`. Use the ids of the view policy and of the pool your platform team created. Here the first attempt names a pool that another Kafka view (`view-6`) already uses: ``` curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X POST "https://$VMS/api/v8/views/" -d '{"path": "/my-events", "bucket": "my-events", "bucket_owner": "my-events-owner", "policy_id": 7, "protocols": ["KAFKA", "S3", "DATABASE"], "kafka_vip_pools": [2], "create_dir": true}' ``` ``` {"detail":"Only MTLS authentication is supported because the Kafka VIP is used by other views: view-6","code":"bad_request"} HTTP 400 ``` With a pool of its own, the view is created. By default it accepts unencrypted connections with no authentication: ``` curl -sk -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X POST "https://$VMS/api/v8/views/" -d '{"path": "/my-events", "bucket": "my-events", "bucket_owner": "my-events-owner", "policy_id": 7, "protocols": ["KAFKA", "S3", "DATABASE"], "kafka_vip_pools": [3], "create_dir": true}' \ | jq "{id, path, bucket, protocols, kafka_vip_pools, tenant_id, kafka_unencrypted_auth_mechanism, is_kafka_unencrypted_conn_allowed}" ``` ``` { "id": 9, "path": "/my-events", "bucket": "my-events", "protocols": [ "KAFKA", "S3", "DATABASE" ], "kafka_vip_pools": [ 3 ], "tenant_id": 3, "kafka_unencrypted_auth_mechanism": "NONE", "is_kafka_unencrypted_conn_allowed": true } ``` For TLS, SASL/PLAIN, mTLS and authorization settings, see [security](/develop/docs/dev-event-broker-operations-security). Decide them before the view is created ([setup](/develop/docs/dev-event-broker-operations-setup#decide-before-the-view-is-created)). ### Create topics `POST /topics/` takes the view's bucket as `database_name`. It returns `201` with an empty body: ``` curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X POST "https://$VMS/api/v8/topics/" -d '{"database_name": "my-events", "name": "orders", "topic_partitions": 4, "retention_ms": 86400000}' ``` ``` HTTP 201 ``` A compacted topic takes `cleanup_policy`, `min_compaction_lag_ms` and `max_message_bytes`. The Terraform provider has no arguments for these, so create compacted topics through the REST API, `vastpy` or the Kafka Admin API: ``` curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X POST "https://$VMS/api/v8/topics/" -d '{"database_name": "my-events", "name": "customers", "topic_partitions": 1, "cleanup_policy": "compact", "min_compaction_lag_ms": 60000, "max_message_bytes": 1048576}' ``` ``` HTTP 201 ``` What compaction does, and its limits, is in [manage topics](/develop/docs/dev-event-broker-guides-topics#compaction) and [topic configuration](/develop/docs/dev-event-broker-reference-topic-configuration). ### List and show topics `GET /topics/` is paginated and returns only names: ``` curl -sk -H "Authorization: Api-Token $VMS_TOKEN" "https://$VMS/api/v8/topics/?database_name=my-events" | jq . ``` ``` { "count": 2, "next": null, "previous": null, "results": [ { "database_name": "my-events", "name": "orders" }, { "database_name": "my-events", "name": "customers" } ] } ``` `GET /topics/show/` returns a topic's settings. Most numbers come back as strings, so compare them as strings in a script: ``` curl -sk -H "Authorization: Api-Token $VMS_TOKEN" "https://$VMS/api/v8/topics/show/?database_name=my-events&name=customers" | jq . ``` ``` { "database_name": "my-events", "name": "customers", "num_events": 0, "capacity": 0, "topic_partitions": 1, "cleanup_policy": "compact", "min_compaction_lag_ms": "60000", "max_message_bytes": "1048576", "message_timestamp_type": "CreateTime", "delete_retention_ms": "86400000", "message_timestamp_after_max_ms": "9223372036854775807", "message_timestamp_before_max_ms": "9223372036854775807", "retention_ms": "604800000" } ``` ### Change a topic `PATCH /topics/` changes settings in place and returns `204`. The partition count is fixed when the topic is created. ``` curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X PATCH "https://$VMS/api/v8/topics/" -d '{"database_name": "my-events", "name": "orders", "retention_ms": 172800000}' ``` ``` HTTP 204 ``` ``` curl -sk -H "Authorization: Api-Token $VMS_TOKEN" "https://$VMS/api/v8/topics/show/?database_name=my-events&name=orders" | jq . ``` ``` { "database_name": "my-events", "name": "orders", "num_events": 0, "capacity": 0, "topic_partitions": 4, "message_timestamp_type": "CreateTime", "retention_ms": "172800000", "cleanup_policy": "delete", "delete_retention_ms": "86400000", "max_message_bytes": "1048588", "message_timestamp_after_max_ms": "9223372036854775807", "message_timestamp_before_max_ms": "9223372036854775807", "min_compaction_lag_ms": "0" } ``` ### Smoke-test the topic End a pipeline by producing to the topic and reading the record back. The script uses confluent-kafka, the supported client ([setup](/develop/docs/dev-event-broker-operations-setup)), with the [producer settings](/develop/docs/dev-event-broker-guides-producing#required-producer-settings) VAST Event Broker needs. It reads the record back from the partition it was written to, so it doesn't create a consumer group. Save it as `kafka_smoke.py`: ``` """Produce one record to a topic and read it back (smoke test after a CI job). usage: python kafka_smoke.py """ import sys, time, uuid from confluent_kafka import Producer, Consumer, TopicPartition bootstrap, topic = sys.argv[1], sys.argv[2] marker = f"ci-smoke-{uuid.uuid4().hex[:8]}" p = Producer({"bootstrap.servers": bootstrap, "acks": "all", "enable.idempotence": False, "compression.type": "none"}) reports = [] p.produce(topic, key=b"smoke", value=marker.encode(), on_delivery=lambda err, msg: reports.append((err, msg))) p.flush(30) if not reports or reports[0][0] is not None: sys.exit(f"produce failed: {reports[0][0] if reports else 'no delivery report'}") md = reports[0][1] print(f"produced {marker} to {md.topic()} partition {md.partition()} offset {md.offset()}") # Read the partition directly from the produced offset: no consumer group is created. c = Consumer({"bootstrap.servers": bootstrap, "group.id": "ci-smoke", "enable.auto.commit": False}) c.assign([TopicPartition(topic, md.partition(), md.offset())]) deadline = time.time() + 15 while time.time() < deadline: rec = c.poll(1.0) if rec is None or rec.error(): continue if rec.value().decode() == marker: print(f"consumed {rec.value().decode()} from partition {rec.partition()} offset {rec.offset()}") break else: c.close() sys.exit("marker not consumed") c.close() ``` ``` python kafka_smoke.py :9092 orders ``` ``` produced ci-smoke-5fb6b031 to orders partition 2 offset 0 consumed ci-smoke-5fb6b031 from partition 2 offset 0 ``` The script exits non-zero if the record doesn't come back, which fails the job. ## vastpy `vastpy` reads the VMS address and token from `VMS_ADDRESS` and `VMS_TOKEN`, and doesn't verify TLS unless you set `VMS_CERT_FILE` ([details](/develop/docs/dev-database-operations-ci-cd)). The same examples use the view `my-events` from the REST section. `vastpy-cli` prints tables, which is handy for a pipeline's log: ``` vastpy-cli get topics database_name=my-events ``` ``` database_name |name --------------+---------------+ my-events |orders my-events |customers ``` ### Keep topics in Git Declare the topics in a YAML file, and let a script create the missing ones and correct drifted settings. It's safe to run on every commit. `topics.yaml`: ``` database_name: my-events topics: - name: orders topic_partitions: 4 retention_ms: 86400000 - name: payments topic_partitions: 2 retention_ms: 259200000 ``` `ensure_topics.py`: ``` """Ensure the Event Broker topics listed in topics.yaml exist (create missing, fix drifted settings). Env: VMS_ADDRESS, VMS_TOKEN (tenant-admin API token).""" import os import yaml from vastpy import VASTClient client = VASTClient(address=os.environ["VMS_ADDRESS"], token=os.environ["VMS_TOKEN"]) spec = yaml.safe_load(open("topics.yaml")) db = spec["database_name"] existing = {t["name"] for t in client.topics.get(database_name=db)["results"]} for want in spec["topics"]: name = want["name"] if name not in existing: client.topics.post(database_name=db, **want) print(f"created {db}/{name}") continue have = client.topics.show.get(database_name=db, name=name) changes = {k: v for k, v in want.items() if k not in ("name", "topic_partitions") and str(have.get(k)) != str(v)} if changes: client.topics.patch(database_name=db, name=name, **changes) print(f"updated {db}/{name}: {changes}") else: print(f"ok {db}/{name}") ``` The script never changes `topic_partitions` (it's fixed at creation), and compares values as strings because `topics/show` returns them as strings. It reads one page of `GET /topics/`; for a view with many topics, follow `next`. It doesn't delete topics that are missing from the file. That's deliberate: deleting a topic deletes its events. The first run creates `payments` and puts back the retention of `orders`, which the REST section changed to two days: ``` python ensure_topics.py ``` ``` updated my-events/orders: {'retention_ms': 86400000} created my-events/payments ``` The second run changes nothing: ``` python ensure_topics.py ``` ``` ok my-events/orders ok my-events/payments ``` ``` vastpy-cli get topics/show database_name=my-events name=payments ``` ``` property |value --------------------------------+--------------------+ database_name |my-events name |payments num_events |0 capacity |0 topic_partitions |2 retention_ms |259200000 message_timestamp_type |CreateTime cleanup_policy |delete delete_retention_ms |86400000 max_message_bytes |1048588 message_timestamp_after_max_ms |9223372036854775807 message_timestamp_before_max_ms |9223372036854775807 min_compaction_lag_ms |0 ``` To keep compacted topics in the same file, add their `cleanup_policy` and related settings; the script passes every field through to the API. ### Create a blob expansion A blob expansion parses each event's JSON value into columns of a target table. **The target table must exist before you create the expansion**, with the same columns. Otherwise the create fails with `404`: ``` vastpy.RESTFailure: response for POST api/latest/blobexpansions with params=None failed with error 404 and message {"detail":"Creating blob_expansion failed. [req=0x5011000588b5]: failed to get target table for blob expansion in @s3_bucket_name=my-events"} ``` So the script creates the schema and table through VMS first. `blob_expansion.py`: ``` """Expand the JSON value of topic orders into a table (blob expansion), then show it. The target table must exist first, so create its schema and table through VMS too. Env: VMS_ADDRESS, VMS_TOKEN (tenant-admin API token).""" import json, os from vastpy import VASTClient client = VASTClient(address=os.environ["VMS_ADDRESS"], token=os.environ["VMS_TOKEN"]) DB, TOPIC, SCHEMA, TARGET = "my-events", "orders", "expanded", "orders_expanded" columns = [ {"name": "order_id", "field": {"column_type": "int64"}}, {"name": "sku", "field": {"column_type": "string"}}, ] client.schemas.post(database_name=DB, name=SCHEMA) client.tables.post(database_name=DB, schema_name=SCHEMA, name=TARGET, arrow_schema=columns) client.blobexpansions.post( database_name=DB, table_name=TOPIC, target_table_schema=SCHEMA, target_table_name=TARGET, expansion_format="json", arrow_schema=columns, ) show = client.blobexpansions.show.get(database_name=DB, table_name=TOPIC) print(json.dumps(show, indent=2)) ``` ``` python blob_expansion.py ``` ``` { "target_table_name": "//my-events/expanded/orders_expanded", "expansion_format": "json", "copy_source_column": false, "source_column_name": "value", "columns": [ { "database_name": "my-events", "schema_name": "kafka_topics", "table_name": "orders", "name": "order_id", "field": { "column_type": "int64" }, "raw_field": "int64" }, { "database_name": "my-events", "schema_name": "kafka_topics", "table_name": "orders", "name": "sku", "field": { "column_type": "string" }, "raw_field": "string" } ], "add_missing_values_output": true, "add_excessive_values_output": true } ``` The expansion is addressed by the topic: `database_name` is the view's bucket and `table_name` is the topic, which is a table in the `kafka_topics` schema. Add a field later with `blobexpansions/add_columns`. The command prints only a blank line: ``` vastpy-cli patch blobexpansions/add_columns database_name=my-events table_name=orders \ arrow_schema='[{"name": "qty", "field": {"column_type": "int32"}}]' ``` ``` vastpy-cli get blobexpansions/show database_name=my-events table_name=orders --json | jq -c ".columns[] | {name, raw_field}" ``` ``` {"name":"order_id","raw_field":"int64"} {"name":"sku","raw_field":"string"} {"name":"qty","raw_field":"int32"} ``` The other settings (`copy_source_column`, `add_missing_values_output` and so on), what happens to events from before the expansion, and how to remove one are in [expand JSON events into columns](/develop/docs/dev-event-broker-guides-blob-expansion). The application that reads the target table with the VAST DataBase SDK needs an identity policy with the `s3:Tabular*` actions on the view's bucket. Owning the bucket isn't enough: without the policy, opening the table fails with `vastdb.errors.Forbidden` (`'code': 'AccessDenied'`, "Permission check failed"). The Terraform example below creates such a user and policy. ## Clean up Remove what the REST and `vastpy` examples created, in this order: the blob expansion, its target table and schema, the topics, the `kafka_topics` schema, and the view. ``` vastpy-cli delete blobexpansions database_name=my-events table_name=orders && echo "blob expansion removed" ``` ``` blob expansion removed ``` Tables and schemas are deleted with a JSON body: ``` curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X DELETE "https://$VMS/api/v8/tables/" -d '{"database_name": "my-events", "schema_name": "expanded", "name": "orders_expanded"}' curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X DELETE "https://$VMS/api/v8/schemas/" -d '{"database_name": "my-events", "name": "expanded"}' ``` ``` HTTP 204 HTTP 204 ``` So are topics. Repeat for each topic: ``` curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X DELETE "https://$VMS/api/v8/topics/" -d '{"database_name": "my-events", "name": "orders"}' ``` ``` HTTP 204 ``` A Kafka view can't be deleted while it holds any schema, and the cluster creates one, `kafka_topics`, in every Kafka view. Deleting all the topics isn't enough: ``` curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -X DELETE "https://$VMS/api/v8/views/9/" ``` ``` {"detail":"Deleting a view with created schemas or tables is not allowed","code":"bad_request"}HTTP 400 ``` Drop the `kafka_topics` schema (and any target schemas of blob expansions), then delete the view: ``` curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -H "Content-Type: application/json" \ -X DELETE "https://$VMS/api/v8/schemas/" -d '{"database_name": "my-events", "name": "kafka_topics"}' curl -sk -w "HTTP %{http_code}\n" -H "Authorization: Api-Token $VMS_TOKEN" -X DELETE "https://$VMS/api/v8/views/9/" ``` ``` HTTP 204 HTTP 204 ``` The VIP pool can't be deleted while a Kafka view uses it (`409`, "Unable to delete … because it is being used by the following Kafka views"). Removing it is your platform team's task. ## Terraform The provider (version 4.0.2) has three resources for Event Broker: `vastdata_view` (the Kafka view), `vastdata_topic` and `vastdata_blob_expansion`. It has none for schemas or tables, so a blob expansion's target table is created outside Terraform. `vastdata_topic` has `topic_partitions`, `retention_ms` and the `message_timestamp_*` settings, but no `cleanup_policy`, compaction settings or `max_message_bytes`. Create compacted topics with the REST API or `vastpy` (above), and keep them out of Terraform so the two don't fight. This configuration creates a Kafka view on a VIP pool your platform team made, its topics from a map with `for_each`, an application user with an identity policy and a key (for reading topic and target tables with the SDK), and, when `expand_orders` is true, a blob expansion of `orders`: ``` # VAST Event Broker: a Kafka view on a VIP pool the platform team provides, its topics, # an application user + identity policy + key (for SDK reads of topic tables), and a blob # expansion of one topic into a table. # Exercised against VAST Cluster 5.5.0-SP6 with provider 4.0.2 as a tenant admin (API token). terraform { required_version = ">= 1.5" required_providers { vastdata = { source = "vast-data/vastdata" version = "4.0.2" } } } # In CI: TF_VAR_vast_host, VASTDATA_API_TOKEN. Provider 4.0.2 needs host in the configuration. variable "vast_host" { type = string } provider "vastdata" { host = var.vast_host skip_ssl_verify = true # lab cluster with a self-signed certificate } variable "tenant_name" { description = "Tenant that holds the Kafka view" type = string } variable "kafka_vip_pool" { description = "Name of a PROTOCOLS VIP pool for this view (from the platform team). A pool already used by another Kafka view only accepts mTLS views." type = string default = "events-ci-pool" } variable "bucket" { type = string default = "events-ci" } variable "topics" { type = map(object({ partitions = number retention_ms = number })) default = { "orders" = { partitions = 4, retention_ms = 86400000 } "payments" = { partitions = 2, retention_ms = 259200000 } } } variable "pgp_public_key" { type = string } variable "expand_orders" { description = "Create the blob expansion. The target table must exist first (no Terraform resource for tables)." type = bool default = false } data "vastdata_tenant" "this" { name = var.tenant_name } data "vastdata_vip_pool" "kafka" { name = var.kafka_vip_pool } resource "vastdata_s3_policy" "events" { name = "${var.bucket}-tabular" tenant_id = data.vastdata_tenant.this.id enabled = true policy = jsonencode({ Version = "2012-10-17" Statement = [{ Effect = "Allow" Action = ["s3:ListBucket", "s3:Tabular*"] Resource = [var.bucket, "${var.bucket}/*"] }] }) } resource "vastdata_user" "app" { name = "${var.bucket}-app" uid = 8205 local_provider_id = data.vastdata_tenant.this.local_provider_id s3_policies_ids = [vastdata_s3_policy.events.id] } resource "vastdata_user_key" "app" { user_id = vastdata_user.app.id tenant_id = data.vastdata_tenant.this.id pgp_public_key = var.pgp_public_key } resource "vastdata_view_policy" "events" { name = "${var.bucket}-policy" tenant_id = data.vastdata_tenant.this.id flavor = "S3_NATIVE" } resource "vastdata_view" "events" { path = "/${var.bucket}" bucket = var.bucket bucket_owner = vastdata_user.app.name protocols = ["KAFKA", "S3", "DATABASE"] policy_id = vastdata_view_policy.events.id tenant_id = data.vastdata_tenant.this.id create_dir = true kafka_vip_pools = [data.vastdata_vip_pool.kafka.id] is_kafka_unencrypted_conn_allowed = true kafka_unencrypted_auth_mechanism = "NONE" } resource "vastdata_topic" "this" { for_each = var.topics database_name = vastdata_view.events.bucket name = each.key topic_partitions = each.value.partitions retention_ms = each.value.retention_ms tenant_id = data.vastdata_tenant.this.id } resource "vastdata_blob_expansion" "orders" { count = var.expand_orders ? 1 : 0 database_name = vastdata_view.events.bucket table_name = vastdata_topic.this["orders"].name target_table_schema = "expanded" target_table_name = "orders" expansion_format = "json" arrow_schema = [ { name = "order_id", field = { column_type = "int64" } }, { name = "sku", field = { column_type = "string" } }, ] tenant_id = data.vastdata_tenant.this.id } output "bootstrap_servers" { value = [for ip in data.vastdata_vip_pool.kafka.ip_ranges : "${ip[0]}:9092"] } output "access_key" { value = vastdata_user_key.app.access_key } output "encrypted_secret_key" { value = vastdata_user_key.app.encrypted_secret_key } ``` The view accepts unencrypted connections without authentication, which suits a lab. For TLS, SASL/PLAIN or mTLS, set the view's `kafka_*` arguments to match what your platform team agreed; see [security](/develop/docs/dev-event-broker-operations-security). `bootstrap_servers` takes the first address of each range in the pool; list three addresses if you have them ([setup](/develop/docs/dev-event-broker-operations-setup)). ### Apply The provider reads the token from `VASTDATA_API_TOKEN`; pass the VMS address as `TF_VAR_vast_host` and the tenant as `TF_VAR_tenant_name`. The user key's secret is encrypted with an ASCII-armored PGP public key. ``` terraform init -no-color | grep -E "Installed|successfully" ``` ``` - Installed vast-data/vastdata v4.0.2 (self-signed, key ID FADBA165459C78A3) Terraform has been successfully initialized! ``` ``` VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" terraform apply -no-color -auto-approve \ | grep -E "Creat|Apply|Error|error|^Outputs|^[a-z_]+ = |:9092" ``` ``` + ":9092", vastdata_s3_policy.events: Creating... vastdata_view_policy.events: Creating... vastdata_s3_policy.events: Creation complete after 1s [name=events-ci-tabular] vastdata_user.app: Creating... vastdata_user.app: Creation complete after 2s [name=events-ci-app] vastdata_user_key.app: Creating... vastdata_user_key.app: Creation complete after 1s vastdata_view_policy.events: Creation complete after 5s [name=events-ci-policy] vastdata_view.events: Creating... vastdata_view.events: Creation complete after 10s [name=view-19] vastdata_topic.this["orders"]: Creating... vastdata_topic.this["payments"]: Creating... vastdata_topic.this["orders"]: Creation complete after 3s [name=orders] vastdata_topic.this["payments"]: Creation complete after 5s [name=payments] Apply complete! Resources: 7 added, 0 changed, 0 destroyed. Outputs: access_key = "" bootstrap_servers = [ ":9092", encrypted_secret_key = < 86400000 # (8 unchanged attributes hidden) } Plan: 0 to add, 1 to change, 0 to destroy. ``` So if Terraform owns a topic, change it in Terraform, not with the Kafka Admin API or the other tools: anything else is undone by the next apply. A scheduled `terraform plan -detailed-exitcode` job is a simple drift alarm. ### Add the blob expansion `vastdata_blob_expansion` fails until its target table exists. The same apply still corrects the drifted topic: ``` VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" TF_VAR_expand_orders=true \ terraform apply -no-color -auto-approve | grep -E "Creat|Modif|Apply|Error|error|│" ``` ``` Error: error creating "blob_expansion" resource. with vastdata_blob_expansion.orders[0], on main.tf line 125, in resource "vastdata_blob_expansion" "orders": 125: resource "vastdata_blob_expansion" "orders" { POST request to https://:443/api/latest/blobexpansions/ returned status code 404 — response body: { "detail": "Creating blob_expansion failed. [req=0x50110006c93e]: failed to get target table for blob expansion in @s3_bucket_name=events-ci" } vastdata_topic.this["orders"]: Modifying... [name=orders] vastdata_topic.this["orders"]: Modifications complete after 0s [name=orders] vastdata_blob_expansion.orders[0]: Creating... ``` Create the schema and the target table with the same columns, here with `vastpy-cli` (or the SDK, or the REST API): ``` vastpy-cli post schemas database_name=events-ci name=expanded && \ vastpy-cli post tables database_name=events-ci schema_name=expanded name=orders \ arrow_schema='[{"name": "order_id", "field": {"column_type": "int64"}}, {"name": "sku", "field": {"column_type": "string"}}]' && \ echo "target table created" ``` ``` target table created ``` Then apply again: ``` VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" TF_VAR_expand_orders=true \ terraform apply -no-color -auto-approve | grep -E "Creat|Modif|Apply|Error|error|│" ``` ``` vastdata_blob_expansion.orders[0]: Creating... vastdata_blob_expansion.orders[0]: Creation complete after 1s Apply complete! Resources: 1 added, 0 changed, 0 destroyed. ``` In a pipeline, put the table step before `terraform apply`, and make it re-runnable (check for the table first, or use the SDK's `create_table(..., fail_if_exists=False)`). To check the expansion, produce two JSON orders and read the target table with the SDK, using the application key from the Terraform outputs. `verify_expansion.py`: ``` """Produce two JSON orders to orders, then read the expanded rows with the VAST DataBase SDK. Env: S3_AK, S3_SK (the application user's key pair). Args: """ import json, os, sys from confluent_kafka import Producer import vastdb bootstrap, endpoint = sys.argv[1], sys.argv[2] p = Producer({"bootstrap.servers": bootstrap, "acks": "all", "enable.idempotence": False, "compression.type": "none"}) for order in ({"order_id": 1001, "sku": "A-1", "qty": 2}, {"order_id": 1002, "sku": "B-7", "qty": 1}): p.produce("orders", value=json.dumps(order).encode()) if p.flush(30): sys.exit("orders not delivered") session = vastdb.connect(endpoint=endpoint, access=os.environ["S3_AK"], secret=os.environ["S3_SK"]) with session.transaction() as tx: t = tx.bucket("events-ci").schema("expanded").table("orders") print([f.name for f in t.columns()]) rows = t.select(columns=["order_id", "sku", "qty"] if "qty" in [f.name for f in t.columns()] else ["order_id", "sku"]).read_all() print(rows.to_pandas().to_string(index=False)) ``` ``` S3_AK=$(terraform output -raw access_key) S3_SK=$(terraform output -raw encrypted_secret_key | gpg --batch --quiet --decrypt) \ python verify_expansion.py :9092 http://:9090 ``` ``` ['order_id', 'sku', 'blob_expansion_key', 'blob_expansion_timestamp_broker', 'blob_expansion_timestamp_producer', 'blob_expansion_headers', 'blob_expansion_error_output', 'blob_expansion_mismatches_output', 'blob_expansion_missing_values_output', 'blob_expansion_excessive_values_output', 'blob_expansion_kafka_partition', 'blob_expansion_external_offset'] order_id sku 1001 A-1 1002 B-7 ``` The cluster added the `blob_expansion_*` metadata columns to the target table; they're described in [expand JSON events into columns](/develop/docs/dev-event-broker-guides-blob-expansion#metadata-columns). `qty` isn't a column because this expansion doesn't declare it. A plan afterwards reports no changes. ### Destroy `terraform destroy` fails on the view while it holds schemas: the target schema you created, and `kafka_topics`, which the cluster creates in every Kafka view. Resources ordered before the view (the key, the blob expansion and the topics) are already gone by then: ``` VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" TF_VAR_expand_orders=true \ terraform destroy -no-color -auto-approve | grep -E "Destroy|destroy|Error|error|│|detail" ``` ``` Error: Delete[view] DELETE request to https://:443/api/latest/views/19/ returned status code 400 — response body: { "detail": "Deleting a view with created schemas or tables is not allowed", "code": "bad_request" } - destroy # vastdata_blob_expansion.orders[0] will be destroyed # vastdata_s3_policy.events will be destroyed # vastdata_topic.this["orders"] will be destroyed # vastdata_topic.this["payments"] will be destroyed # vastdata_user.app will be destroyed # vastdata_user_key.app will be destroyed # vastdata_view.events will be destroyed # vastdata_view_policy.events will be destroyed Plan: 0 to add, 0 to change, 8 to destroy. vastdata_user_key.app: Destroying... vastdata_blob_expansion.orders[0]: Destroying... vastdata_topic.this["payments"]: Destroying... [name=payments] vastdata_topic.this["orders"]: Destroying... [name=orders] vastdata_view.events: Destroying... [name=view-19] ``` Drop the target table and both schemas, then run destroy again: ``` vastpy-cli delete tables database_name=events-ci schema_name=expanded name=orders && \ vastpy-cli delete schemas database_name=events-ci name=expanded && \ vastpy-cli delete schemas database_name=events-ci name=kafka_topics && echo "schemas dropped" ``` ``` schemas dropped ``` ``` VASTDATA_API_TOKEN=$VMS_TOKEN TF_VAR_pgp_public_key="$(cat ci.pub.asc)" TF_VAR_expand_orders=true \ terraform destroy -no-color -auto-approve | grep -E "Destroy|Destruction|Error|error|│|detail" ``` ``` vastdata_view.events: Destroying... [name=view-19] vastdata_view.events: Destruction complete after 5s vastdata_user.app: Destroying... [name=events-ci-app] vastdata_view_policy.events: Destroying... [name=events-ci-policy] vastdata_view_policy.events: Destruction complete after 1s vastdata_user.app: Destruction complete after 4s vastdata_s3_policy.events: Destroying... [name=events-ci-tabular] vastdata_s3_policy.events: Destruction complete after 0s Destroy complete! Resources: 4 destroyed. ``` Destroying topics deletes their events. Guard destroy jobs with a manual approval, and consider `lifecycle { prevent_destroy = true }` on topics that hold data you need. ## VAST admin MCP server The VAST admin MCP server (`vast-admin-mcp`) has no tools for topics, blob expansions or brokers: none of its tools mentions them. It can list views, and so show which ones are Kafka views (`protocols` contains `KAFKA`), with `list_views_vast`. This output was filtered on the client to the Kafka views and three of their fields: ``` call list_views_vast {"path": "/*"} isError: False [ { "path": "/events-ci", "bucket": "events-ci", "protocols": "S3,DATABASE,KAFKA" }, { "path": "/my-events", "bucket": "my-events", "protocols": "S3,DATABASE,KAFKA" } ] ``` It can't create a Kafka view: its `create_view_vast` tool has no VIP pool argument, and the cluster refuses the view without one. ``` call create_view_vast {"cluster": "", "tenant": "", "path": "/my-events-mcp", "protocols": "KAFKA,S3,DATABASE", "bucket": "my-events-mcp", "bucket_owner": "my-app", "policy": "my-db-policy"} isError: True Error calling tool 'create_view_vast': response for POST api/latest/views with params=None failed with error 400 and message {"detail":"VIP pools must be provided for a view with the KAFKA protocol","code":"bad_request"} ``` For an assistant that works with topics, give it a Kafka client and the Kafka Admin API instead. Setup and the server's other tools are in [CI/CD and automation for VAST DataBase](/develop/docs/dev-database-operations-ci-cd). ## A CI job This GitLab CI job applies `topics.yaml` with `ensure_topics.py` on every push to the default branch, then smoke-tests one topic. **It was not run end to end on a CI runner.** Each step it runs was run by hand on a 5.5.0-SP6 lab cluster, as shown above. ``` # Not run end to end on GitLab. The steps match the vastpy and smoke-test sections of this page. # # CI/CD variables (Settings > CI/CD > Variables): # VMS_ADDRESS VMS address # VMS_TOKEN tenant-admin API token (masked, protected) # KAFKA_BOOTSTRAP :9092 of the view in topics.yaml # The runner must reach VMS on port 443 and every VIP in the Kafka view's pool. stages: [topics, verify] ensure-topics: stage: topics image: python:3.12-slim resource_group: event-broker-topics # one run at a time rules: - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH changes: [topics.yaml] before_script: - pip install vastpy==0.3.22 pyyaml script: - python ensure_topics.py smoke-test: stage: verify image: python:3.12-slim needs: [ensure-topics] rules: - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH changes: [topics.yaml] before_script: - pip install confluent-kafka==2.8.0 script: - python kafka_smoke.py "$KAFKA_BOOTSTRAP" orders ``` For a Terraform-based job, run `terraform plan` on merge requests and `terraform apply` on the default branch, with `VASTDATA_API_TOKEN` as a masked variable and remote state; [CI/CD and automation for VAST DataBase](/develop/docs/dev-database-operations-ci-cd) covers state and secrets. ## See also - [CI/CD and automation for VAST DataBase](/develop/docs/dev-database-operations-ci-cd): tokens, TLS, users, keys and Terraform state - [CI/CD and automation for VAST Vector Search](/develop/docs/dev-vector-search-operations-ci-cd) and [for VAST DataEngine](/develop/docs/dev-dataengine-operations-ci-cd) - [Automation tools compared](/develop/docs/dev-database-reference-automation-tools) - [Setup](/develop/docs/dev-event-broker-operations-setup) and [security](/develop/docs/dev-event-broker-operations-security) - [Manage topics](/develop/docs/dev-event-broker-guides-topics), [topic configuration](/develop/docs/dev-event-broker-reference-topic-configuration) and [expand JSON events into columns](/develop/docs/dev-event-broker-guides-blob-expansion)