Setup

Applies to VAST Cluster 5.5.

Your platform team creates the broker: a Kafka view, its VIP pool, and its security settings. This page lists what to ask them for, gives you a request you can copy, and shows how to check that your Kafka client and the VAST DataBase SDK can both reach it.

What you need

Admin task: send to your platform team. The setup steps are in the KB:

• Configuring VAST Event Broker: the VIP pool, the bucket owner and the Kafka view.
• Configuring user authentication: TLS and SASL/PLAIN, and the cluster's Kafka TLS certificate.
• Configuring mTLS authentication (5.5).
• Configuring user authorization.
• Granting Kafka permissions in an identity policy.
• Tenant administrators: the tenant guide version. The VIP pool needs a cluster administrator.

What your platform team sets up for a broker, in order, before your Kafka client connects.

Decide before the view is created

Some choices can't be changed later, so agree on them first:

• The VIP pool. A Kafka view has exactly one VIP pool, with at least one virtual IP per CNode. Once the view exists, the pool can be modified but not replaced by another one.
• Kafka on a new view. The Kafka protocol can't be added to an existing view, so the broker is always a new view. It can't be removed from a view that holds topics either.
• Partition counts. Each topic's partition count is fixed when the topic is created, up to 20,000 per topic and 200,000 per view. Size it for the most consumers you'll run in one group.
• Security. Whether clients connect with TLS, and how they authenticate. If you want the topics protected per user, ask for authorization to be turned on. Without it, any client that can connect can read, write and delete every topic.

If nothing is set, a new Kafka view accepts unencrypted connections with no authentication, and doesn't check permissions.

What to send your platform team

Copy this, and fill in the parts in angle brackets:

Please set up a VAST Event Broker (Kafka) view for <project>:

- Kafka view on a new path, tenant <tenant>, with its own VIP pool
  (role Protocols, at least one VIP per CNode).
- Connections: <TLS on port 9093 | unencrypted on port 9092 | both>.
- Authentication: <none | SASL/PLAIN | mTLS>.
  - SASL/PLAIN: a VAST local user <name> and its password.
  - mTLS: a client certificate for user <name>, and the CA certificate.
- Authorization: <on | off>. If on, an identity policy for <user> that allows
  KafkaRead and KafkaWrite on topics <topics>, KafkaRead on consumer groups
  <groups>, and <KafkaCreate / KafkaDelete if we manage topics ourselves>.
- Topics: <names, partition counts, retention>, or permission to create them.
- For SQL/SDK reads of the topics: an S3 access key pair, the database
  endpoint, and read access to the view's bucket.
- For blob expansion: s3:Tabular*BlobExpansion actions on the bucket, and a
  schema <schema> we can create tables in.

Please send back: the bucket name, three bootstrap VIPs, the port(s), the CA
certificate (for TLS), and the credentials.

Client configuration

Every client needs the bootstrap addresses. Producers also need two settings for VAST Event Broker, whatever their language: no compression, and no idempotence.

import os

conf = {
    "bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"],  # e.g. "<vip1>:9092,<vip2>:9092,<vip3>:9092"
}

producer_conf = {
    **conf,
    "compression.type": "none",     # compressed batches are rejected
    "enable.idempotence": False,    # idempotent producing isn't supported
}

Set both explicitly in every producer, even where they're your client's default. The settings for TLS, SASL/PLAIN and mTLS are on the security page.

Check that it works

These checks read and create nothing. Set the environment variables first:

export KAFKA_BOOTSTRAP="<vip1>:9092,<vip2>:9092,<vip3>:9092"
# Only if the view uses SASL/PLAIN:
export KAFKA_USERNAME="<local user>"
export KAFKA_PASSWORD="<password>"

The Kafka connection

This asks the broker for its metadata. It works on the unencrypted port, with or without SASL/PLAIN. For TLS, add the settings from security.

import os
from confluent_kafka.admin import AdminClient

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"],
    })

admin = AdminClient(conf)
metadata = admin.list_topics(timeout=10)
brokers = list(metadata.brokers.values())
bootstrap = os.environ["KAFKA_BOOTSTRAP"].split(",")
print("brokers:", len(brokers), "on port(s)", sorted({b.port for b in brokers}))
print("every bootstrap address is a broker:",
      all(any(f"{b.host}:{b.port}" == addr for b in brokers) for addr in bootstrap))
print("topics:", len(metadata.topics))

brokers: 3 on port(s) [9092]
every bootstrap address is a broker: True
topics: 0

You should see one broker per virtual IP in the pool. If the call times out, check the address and port, and that your network reaches the virtual IPs. If the connection closes straight away, check the authentication settings; see troubleshooting.

The topic tables

This lists the topics as tables, through the VAST DataBase SDK. It needs the S3 key pair and database endpoint, and the bucket name of the Kafka view:

export AWS_ACCESS_KEY_ID="<access key>"
export AWS_SECRET_ACCESS_KEY="<secret key>"
export AWS_S3_ENDPOINT_URL="http://<database endpoint>"
export EVENTS_BUCKET="<bucket of the Kafka view>"

import os
import vastdb

session = vastdb.connect()
with session.transaction() as tx:
    schema = tx.bucket(os.environ["EVENTS_BUCKET"]).schema("kafka_topics")
    print("topic tables:", sorted(schema.tablenames()))

topic tables: []

The list matches the topics from the Kafka check, if your Kafka user can see all of them.

See also

• Security: authentication modes, client settings and permissions.
• Quickstart: produce, consume and query a topic.
• Topics: create topics with the Kafka Admin API.
• VAST DataBase setup prerequisites: keys, endpoint and Tabular permissions for the SDK.