Manage topics

Applies to VAST Cluster 5.5.

This guide shows how to create, inspect, change and delete VAST Event Broker topics from a Kafka admin client: partitions, retention, compaction and the other topic settings. It also says when to use the VAST Web UI or CLI instead, and what the broker rejects.

Run the code blocks in order in one Python session. Below each block is the output you should see.

Ways to create a topic

There's no automatic topic creation: producing to a topic that doesn't exist fails (Kafka Protocol Support). You also can't create a topic as a table with the VAST DataBase SDK; the cluster refuses it.

Topics live in the Kafka view's bucket, in the schema kafka_topics, one table per topic. See query topics.

Set up

You need the environment variables from the quickstart. If authorization is turned on for the view, creating a topic needs KafkaCreate, changing its settings needs KafkaAlterConfigs, and deleting it needs KafkaDelete. See security.

import os

from confluent_kafka import KafkaException
from confluent_kafka.admin import (AdminClient, AlterConfigOpType, ConfigEntry, ConfigResource,
                                  NewPartitions, NewTopic)

ORDERS = "docs-topics-orders"
PROFILES = "docs-topics-profiles"

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 wait(futures):
    """Wait for each admin operation and return its result or the Kafka error name."""
    results = {}
    for key, future in futures.items():
        name = key.name if isinstance(key, ConfigResource) else key
        try:
            results[name] = future.result()
        except KafkaException as e:
            results[name] = e.args[0].name()
    return results

admin = AdminClient(kafka_conf())
print("connected:", len(admin.list_topics(timeout=10).brokers) > 0)

connected: True

Create a topic

Give the partition count and any settings you want to change from the defaults. This topic keeps events for one day:

day_ms = 24 * 60 * 60 * 1000

print(wait(admin.create_topics([
    NewTopic(ORDERS, num_partitions=3, replication_factor=1,
             config={"retention.ms": str(day_ms), "cleanup.policy": "delete"}),
], operation_timeout=30)))

topic = admin.list_topics(ORDERS, timeout=10).topics[ORDERS]
print("partitions:", len(topic.partitions))

{'docs-topics-orders': None}
partitions: 3

A result of None means the topic was created.

Choose the partition count

A partition is the unit of ordering and of parallel reads: in a consumer group, each partition is read by one consumer at a time. Choose the count with care:

• You can't change it after the topic is created (Kafka Protocol Support). To get more partitions, create a new topic and move your producers and consumers to it.
• A topic can have up to 20,000 partitions, and a Kafka-enabled view up to 200,000 in total.
• Pass an explicit count. A count of -1 (the broker default) is rejected.

replication_factor has no effect: the events are stored once in VAST DataBase, and the cluster protects them. Pass 1. Explicit replica assignments are rejected.

Topic names

Use letters, digits, ., _ and -. A topic name is also the table name in kafka_topics, so a name like orders.v1 or click-stream works in Kafka but needs quoting in SQL.

Validate without creating

Pass validate_only=True to check a request without creating anything:

print(wait(admin.create_topics([
    NewTopic("docs-topics-dry-run", num_partitions=6, replication_factor=1,
             config={"cleanup.policy": "compact"}),
], validate_only=True, operation_timeout=30)))
print("exists:", "docs-topics-dry-run" in admin.list_topics(timeout=10).topics)

{'docs-topics-dry-run': None}
exists: False

What the broker rejects

These are the errors for the common mistakes:

bad_requests = {
    "already exists":       NewTopic(ORDERS, num_partitions=3, replication_factor=1),
    "too many partitions":  NewTopic("docs-topics-huge", num_partitions=20_001, replication_factor=1),
    "invalid name":         NewTopic("docs topics/bad name", num_partitions=1, replication_factor=1),
    "size-based retention": NewTopic("docs-topics-bytes", num_partitions=1, replication_factor=1,
                                     config={"retention.bytes": "1073741824"}),
    "unknown setting":      NewTopic("docs-topics-unknown", num_partitions=1, replication_factor=1,
                                     config={"no.such.setting": "1"}),
}
for label, new_topic in bad_requests.items():
    try:
        result = wait(admin.create_topics([new_topic], operation_timeout=30))[new_topic.topic]
    except (KafkaException, ValueError) as e:
        result = f"rejected by the client: {e}"
    print(f"{label:21} {result}")

already exists        TOPIC_ALREADY_EXISTS
too many partitions   INVALID_PARTITIONS
invalid name          TOPIC_EXCEPTION
size-based retention  INVALID_CONFIG
unknown setting       INVALID_CONFIG

There's no size-based retention: retention.bytes isn't a valid setting.

Adding partitions isn't supported, so the client refuses the request:

try:
    print(wait(admin.create_partitions([NewPartitions(ORDERS, 6)], operation_timeout=30)))
except KafkaException as e:
    print(e.args[0].name(), "-", e.args[0].str())

{'docs-topics-orders': '_UNSUPPORTED_FEATURE'}

The Confluent client checks which APIs the broker offers before it sends a request. The broker doesn't offer CreatePartitions, so the client fails with _UNSUPPORTED_FEATURE without sending anything.

Topic settings

These settings take effect. You can set them at creation, or change them later with alter_configs:

Many other standard Kafka topic settings are accepted and stored but have no effect, and an unknown name is rejected. The topic configuration reference lists every setting and what happens to it. In the VAST Web UI, max.message.bytes is the topic's Max size, in MB: above 5 MB, the topic can't be compacted or queried as a table (KB).

Read a topic's settings

shown = ["retention.ms", "cleanup.policy", "max.message.bytes", "message.timestamp.type"]

def show_configs(topic):
    configs = wait(admin.describe_configs([ConfigResource("TOPIC", topic)]))[topic]
    for name in shown:
        print(f"{name:24} {configs[name].value}")

show_configs(ORDERS)

retention.ms             86400000
cleanup.policy           delete
max.message.bytes        1048588
message.timestamp.type   CreateTime

Change a topic's settings

Use alter_configs. VAST Event Broker supports the Kafka AlterConfigs request, not IncrementalAlterConfigs. In Kafka, AlterConfigs replaces the topic's whole set of settings, so include every setting you want to keep, not just the one you're changing:

print(wait(admin.alter_configs([
    ConfigResource("TOPIC", ORDERS, set_config={
        "retention.ms": str(3 * day_ms),
        "cleanup.policy": "delete",
        "message.timestamp.type": "LogAppendTime",
    }),
])))
show_configs(ORDERS)

{'docs-topics-orders': None}
retention.ms             259200000
cleanup.policy           delete
max.message.bytes        1048588
message.timestamp.type   LogAppendTime

Changes to retention and compaction apply to the events already in the topic, not only to new ones. incremental_alter_configs, which newer Kafka tools use by default, fails:

try:
    print(wait(admin.incremental_alter_configs([
        ConfigResource("TOPIC", ORDERS, incremental_configs=[
            ConfigEntry("retention.ms", str(day_ms), incremental_operation=AlterConfigOpType.SET),
        ]),
    ])))
except KafkaException as e:
    print(e.args[0].name(), "-", e.args[0].str())

{'docs-topics-orders': '_UNSUPPORTED_FEATURE'}

The same check applies: the broker doesn't offer IncrementalAlterConfigs, so use alter_configs().

You can't rename a topic, or change its partition count, through any interface.

Retention

Retention removes old events; compaction keeps the latest event per key.

With cleanup.policy=delete, a background job removes events once they're older than retention.ms:

• Age is measured from the time the broker received the event, not the producer's timestamp.
• Deletion isn't immediate. Expect expired events to stay readable for a while after they pass the retention period. Don't rely on retention to hide data from consumers.
• The newest event in a partition is never removed by retention, so a partition that stops receiving events keeps its last one.
• Retention moves the partition's earliest offset forward. A consumer whose committed offset has expired falls back to its auto.offset.reset setting.

Compaction

With cleanup.policy=compact, the cluster keeps only the latest event for each key in a partition, and deletes earlier ones (KB). Use it for topics that hold the current state of something, such as a profile per user. Create a compacted topic like this:

print(wait(admin.create_topics([
    NewTopic(PROFILES, num_partitions=1, replication_factor=1, config={
        "cleanup.policy": "compact",
        "min.compaction.lag.ms": str(60 * 60 * 1000),
        "delete.retention.ms": str(day_ms),
    }),
], operation_timeout=30)))
show_configs(PROFILES)

{'docs-topics-profiles': None}
retention.ms             604800000
cleanup.policy           compact
max.message.bytes        1048588
message.timestamp.type   CreateTime

What to expect:

• Compaction runs in the background. Until it has run, consumers and queries can see several events with the same key.
• Every event on a compacted topic needs a key.
• To delete a key, produce a tombstone: the key with a null value. The tombstone itself is removed after delete.retention.ms.
• Offsets don't change when compaction removes events, so a compacted partition has gaps in its offsets, as in Kafka.
• A topic whose Max size is above 5 MB can't be compacted.

Delete a topic

Deleting a topic deletes all its events, and the committed offsets of every consumer group for it:

print(wait(admin.delete_topics([ORDERS, PROFILES], operation_timeout=30)))
remaining = admin.list_topics(timeout=10).topics
print("still listed:", [t for t in (ORDERS, PROFILES) if t in remaining])

{'docs-topics-orders': None, 'docs-topics-profiles': None}
still listed: []

Dropping the topic's table in kafka_topics with the VAST DataBase SDK or SQL also deletes the topic. If authorization is on, the Kafka delete needs KafkaDelete on the topic.

Clean up

The previous section deleted both topics. Delete anything else this page created, in case a step above behaved differently on your cluster, and check that nothing is left:

leftover = [t for t in admin.list_topics(timeout=10).topics if t.startswith("docs-topics-")]
if leftover:
    wait(admin.delete_topics(leftover, operation_timeout=30))
print("leftover topics:", [t for t in admin.list_topics(timeout=10).topics if t.startswith("docs-topics-")])

leftover topics: []

See also

• Topic configuration: every topic setting, and whether it's honoured, accepted and ignored, or rejected.
• Producing and consuming.
• Concepts: topics, partitions, retention and compaction.
• Limits.
• KB: Managing Event Topics, and the CLI reference for topic create.