Publishing Events to VAST Event Broker

VAST Event Broker collects events occurring to elements in VAST Element Store into Kafka event topics. VAST Event Broker is based on the VAST Database which allows consumers to query event topics via the various database APIs.

VAST Event Broker utilizes VAST proprietary implementation of the Kafka protocol, as detailed in Kafka Protocol Support.

The workflow is as follows:

The VAST implementation of the Kafka protocol supports a basic subset of the Kafka APIs to allow clients to publish and consume events from the VAST Event Broker.

VAST Event Broker supports:

The following Kafka capabilities are not supported:

The following limitations apply:

VAST Cluster supports Confluent Kafka Python client 2.4 - 2.8.

Note

Since VAST Event Broker requires a dedicated VIP pool, which must be designated from the available virtual IPs on the cluster, this configuration requires a ​cluster admin​​ manager user. If you want to provision of a VAST Event Broker for a tenant, a cluster admin user must create the VAST Event Broker view for the tenant.

Complete these steps to configure VAST Event Broker:

With you cluster admin, verify that there is a virtual IP pool that can be used to access the event topics using the Kafka protocol. The pool must have the PROTOCOLS roles assigned.

If the VAST Event Broker view will be associated with a view policy that includes virtual IP pools, the pool specified as the Kafka pool must be one of the view policy pools.

Configure a user on the authentication provider attached to the tenant where VAST Event Broker will run so that the user has S3 bucket creation permissions and S3 access keys.

This user will be assigned as the bucket owner when creating a view for VAST Event Broker.

VAST Event Broker runs on a view that has Kafka protocol enabled.

Note

View nesting is not supported.

Follow the steps in Creating Views to create a VAST Event Broker view as follows:

(Optional) Configure authentication of users accessing the Kafka-enabled view.

Users accessing a Kafka-enabled view can be authenticated against the VAST provider associated with the tenant using the PLAIN Simple Authentication and Security Layer (SASL) mechanism. SASL/PLAIN authentication is available on both encrypted (TLS) and non-encrypted connections.

To configure user authentication:

If you are going to use SASL/PLAIN authentication with TLS, ensure that the VAST cluster has been configured with a TLS certificate intended for use with Kafka-enabled views. To check or upload a Kafka TLS certificate and a key:

The following limitations apply:

(Optional) Configure mutual TLS (mTLS) authentication between VAST Event Broker and event consumers, which allows the server to authenticate clients connecting to a Kafka-enabled view on the VAST cluster.

In multi-tenant environments, the VAST cluster can be configured to distinguish between mTLS certificates associated with particulars tenants, and thus to direct the IO to the intended tenant based on the mTLS certificate (instead of using the client IP or destination virtual IP for that purpose). This improves security while enforcing tenant isolation and tenant-level authorization for NFS clients.

Warning

If your cluster has a cluster-wide configuration of mTLS authentication for Kafka connections, Kafka mTLS certificates configured at the tenant level are not used, and the cluster authenticates Kafka clients based on the cluster-wide settings only.

If mTLS authentication is enabled, connections without a proper CA certificate are rejected.

The CA certificate can be tenant-specific or global (cluster-wide, accepted by all tenants).

Note

With VAST Event Broker mTLS in place, VAST Event Broker views hosted on multiple tenants can use the same virtual IP pool, since determination of the correct tenant for each client request will be done based on the information in the CA certificate.

You configure mTLS authentication for Kafka by adding an mTLS certificate in tenant settings.

After adding an mTLS certificate, you need to enable mTLS authentication for the VAST Event Broker view.

A tenant-level mTLS certificate must be unique across all tenants.

If the tenant is configured with a CA certificate and the client sends a certificate that is not signed by this CA (or a trust chain that does not provide for the signing), access is denied.

To add an mTLS certificate for a tenant:

To enable mTLS authentication for a VAST Event Broker view:

(Optional) Configure authorization of users accessing the Kafka-enabled view.

Only explicitly allowed operations can be performed.

You create, view, modify and delete event topics similar to VAST Database tables. When you create a VAST Event Broker view, VAST Cluster automatically creates an associated VAST database with a schema labelled Kafka-Compatible Broker Topics. Under this schema, you can create and manage topics where events will be recorded.

To create a topic in VAST Web UI:

To modify a topic in VAST Web UI:

You grant permissions for Kafka-related objects by listing operations that a user is allowed or prohibited to perform in the user's identity policy.

By default, if no ACL is found, the operation is allowed for any user. If you want to alter this behavior, contact VAST Support.

The following Kafka operations are supported:

For example, the following identity policy allows the user to read and write to the topic, but prohibits topic creation and deletion:

"Version": "2012-10-17",
"Statement": [
  {
    "Sid": "Aaa__gggg",
    "Action": "KafkaRead",
    "Effect": "Allow",
    "Resource": "vast:kafka:topic:kafka-pppp:ddd_0"
  },
  {
    "Sid": "Aaa__ssss",
    "Action": "KafkaWrite",
    "Effect": "Allow",
    "Resource": "vast:kafka:topic:kafka-pppp:ddd_0"
  },
  {
    "Sid": "Aaa__dddd",
    "Action": "KafkaCreate",
    "Effect": "Deny",
    "Resource": "vast:kafka:topic:kafka-pppp:ddd_0"
  },
  {
    "Sid": "Aaa__cccc",
    "Action": "KafkaDelete",
    "Effect": "Deny",
    "Resource": "vast:kafka:topic:kafka-pppp:ddd_0"
  },
]

Prior to using Kafka APIs, ensure that the prerequisites described in Configuring VAST Event Broker are met.

The following examples use a Kafka client that can be installed with:  pip install kafka-python.

In the examples, you need to supply a list of Kafka bootstrap server IPs. These are the IPs that a Kafka client uses to connect to the cluster. It is recommended to specify three of the virtual IPs from the virtual IP pool assigned to the VAST Event Broker view.

from kafka.admin import KafkaAdminClient, NewTopic
BOOSTRAP_SERVERS = "<comma-separated list of IPs>"
topic_name = #
# Kafka Admin Configuration
admin_client = KafkaAdminClient(
    bootstrap_servers=[BOOSTRAP_SERVERS],
    client_id='python-admin-client'
)
topic_name = topic_name
num_partitions = 2
replication_factor = 1
topic = NewTopic(
    name=topic_name,
    num_partitions=num_partitions,
    replication_factor=replication_factor
)
try:
    # Create the topic
    admin_client.create_topics(new_topics=[topic], validate_only=False)
    print(f"Topic '{topic_name}' created successfully with {num_partitions} partitions and replication factor {replication_factor}.")
except Exception as e:
    print(f"Error creating topic: {e}")
finally:
    admin_client.close()

BOOTSTRAP_SERVER = "<comma-separated list of IPs>"
from kafka.admin import KafkaAdminClient
# Define the Kafka broker (adjust the host/port as needed)
# Create an admin client
admin_client = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVER)
# Fetch the list of topics
topics = admin_client.list_topics()
# Print the list of topics
print("Kafka Topics:", topics)
# Close the admin client
admin_client.close()

from kafka import KafkaProducer
import json
BOOSTRAP_SERVERS = "<comma-separated list of IPs>"
producer = KafkaProducer(bootstrap_servers=BOOSTRAP_SERVERS)
for i in range (100,2500):
    key = str(i)
    message = {"name": "vast", "id": i}
    producer.send(topic_name, key=key.encode("utf-8") , value=json.dumps(message).encode('utf-8'))
producer.flush()
print("Message sent to Kafka topic")

BOOSTRAP_SERVERS = "<comma-separated list of IPs>"
from kafka import KafkaConsumer
consumer = KafkaConsumer(
    topic_name,
    bootstrap_servers=[BOOSTRAP_SERVERS],
    auto_offset_reset='earliest',
    group_id='your-consumer-group-name'
)
for message in consumer:
    print(f"Received message: {message.value.decode('utf-8')}")

from kafka.admin import KafkaAdminClient, NewTopic
# Define the Kafka broker
BOOTSTRAP_SERVER = "<comma-separated list of IPs>"
TOPIC_TO_DELETE = "my_topic"  # Change this to the topic you want to delete
# Create an admin client
admin_client = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVER)
# Delete the topic
try:
    admin_client.delete_topics([TOPIC_TO_DELETE])
    print(f"Topic '{TOPIC_TO_DELETE}' deleted successfully.")
except Exception as e:
    print(f"Failed to delete topic '{TOPIC_TO_DELETE}': {e}")
# Close the admin client
admin_client.close()

VAST Event Broker topics can be queried through the VAST Database API, and also using VAST Database connectors (e.g. the connectors for Trino and Spark).

The following Trino example shows the schema (kafka_topics) of the VAST Database used to store event topics, among other schemas on the VAST cluster:

trino> show schemas in vast;
                     Schema                      
-------------------------------------------------
 information_schema                              
 kafka2/kafka_topics                             
 system_schema                                   
 vast-audit-log-bucket/vast_audit_log_schema     
 vast-big-catalog-bucket/vast_big_catalog_schema 
(5 rows)

The following Trino example shows topics within the kafka_topics schema:

trino> show tables in vast."kafka2/kafka_topics";
         Table          
------------------------
 docker-connect-configs 
 docker-connect-offsets 
 docker-connect-status  
 my-new-topic           
 myfirsttopic           
 newtopic               
 streaming-topic                     
(7 rows)

VAST Cluster provides analytic reports that show latency and bandwidth for produced and fetched events and messages per VAST Event Broker view and per CNode.

To access Kafka-related analytics in VAST Web UI, go to Analytics -> Analytics -> Predefined Analytics tab, set the View or CNode filter and select one of the reports containing the word Kafka.

The following data protection capabilities are supported for VAST Event Broker topics:

The following rules and limitations apply:

Information in Kafka event messages can be parsed and saved in a VAST database table that can then be queried using any of the query methods available for VAST database tables.

You configure parsing by creating or selecting a VAST database table that will store parsed Kafka event data, and then selecting the specific fields to be parsed from the Kafka events. Columns corresponding to the these fields are added to the table (if not already present), and all subsequent events are parsed for these fields.

You can view statistics for parsed Kafka events using custom analytics in the Analytics page of the VAST Web UI.

In order to configure or modify event parsing configurations in the VAST Database, these S3 permissions are required:

These are described in Managing Permissions for Accessing VAST Tabular Databases

This example illustrates how parsing can be configured for a Kafka topic, using the VAST PySDK.

import pyarrow as pa
import vastdb  # pip install vastdb  (requires Python 3.10)
from vastdb.table import BlobExpansionConfig, ExpansionFormat

# Connection info
kafka_bucket  = "my_kafka_bucket"
kafka_schema  = "kafka_topics"      # the schema Kafka topics live under
topic_name    = "my_topic"          # source table = Kafka topic name
target_schema = "expanded_data"     # different schema, same bucket
target_table  = "my_expanded_table"

# NOTE: the source table (topic_name) and its Kafka topic must be created beforehand

# 1. Connect
session = vastdb.connect(access_key, secret_key, endpoint)

# 2. Create the target schema (different schema, same bucket)
with session.transaction() as tx:
    tx.bucket(kafka_bucket).create_schema(target_schema)

# 3. Define what the expanded blob looks like
expansion_schema = pa.schema([
    pa.field("event_type", pa.string()),
    pa.field("id",  pa.int64()),
    # ... any fields the JSON/blob contains
])

# 4. Create the target table in the target schema
with session.transaction() as tx:
    tx.bucket(kafka_bucket).schema(target_schema).create_table(
        target_table, expansion_schema, sorting_key=[]
    )

# 5. Create the blob expansion (links source blob column → target table)
with session.transaction() as tx:
    be = tx.bucket(kafka_bucket).schema(kafka_schema).table(topic_name).create_blob_expansion(
        expansion_schema=expansion_schema,
        target_table_name=target_table,
        target_table_schema=target_schema,   # required: target is in a different schema
        source_column_name="value",
        config=BlobExpansionConfig(
            expansion_format=ExpansionFormat("json"),
            copy_source_column=False,
            flatten_path=False,
            flatten_delimiter="__",
        ),
    )

# 6. Add a column to the blob expansion
columns_to_add = pa.schema([
    pa.field("user_id", pa.string()),
])
with session.transaction() as tx:
    be = tx.bucket(kafka_bucket).schema(kafka_schema).table(topic_name).blob_expansion("value")
    be.add_columns(columns_to_add=columns_to_add)

# 7. Remove a column from the blob expansion
columns_to_remove = pa.schema([
    pa.field("id", pa.int64()),
])
with session.transaction() as tx:
    be = tx.bucket(kafka_bucket).schema(kafka_schema).table(topic_name).blob_expansion("value")
    be.drop_columns(columns_to_remove=columns_to_remove)

# 8. Drop the blob expansion entirely
with session.transaction() as tx:
    tx.bucket(kafka_bucket).schema(kafka_schema).table(topic_name).blob_expansion("value").drop()