Virtual Kafka Clusters

A virtual Kafka cluster gives a team its own Kafka endpoint on top of a shared Kafka cluster. It claims part of the backing cluster’s capacity, in Kafka Units, and enforces produce and fetch limits for it. Clients use it like any Kafka cluster.

Key Features

Many teams, one Kafka cluster: each virtual cluster has its own endpoint and its own share of the capacity.
Capacity in Kafka Units: the virtual clusters on a Kafka cluster together can’t claim more than its Cluster Capacity (Kafka Units).
Produce and fetch limits follow the capacity it claims.
OAuth sign-in for clients, with your identity provider’s issuer, audience and keys.

Settings

Basic Configuration: Identifier and Provider.
Kafka Cluster Selection: the backing Kafka Cluster, and the Capacity Allocation (Kafka Units) to claim from it.
Advanced Configuration: client authentication (the OAuth Issuer, Audience, JWKS Endpoint, Authorization URI, Token URI and Scope).

A virtual cluster belongs to the fleet of its Kafka cluster, so the API addresses it under that fleet: $API/fleet/<fleet>/kafka/kvc/.

Before you start

A fleet in Healthy state with room for the Virtual Kafka Cluster workload (see Fleet management).
For the API: an access token in STREAMTIME_TOKEN and your organization identifier in place of <org>. Replace api.example.com with your Streamtime API address.

The API examples below share this setup:

HTTPPythonTypeScriptGo

export STREAMTIME_TOKEN=<your-access-token>
export API=https://api.example.com/organizations/<org>

import os
from streamtime_sdk.client import AuthenticatedClient

client = AuthenticatedClient(
    base_url="https://api.example.com",
    token=os.environ["STREAMTIME_TOKEN"],
)
ORG = "<org>"

Uses the generated client in streamtime-sdk/typescript/client.

import { client } from "./client/client.gen";

client.setConfig({
  baseUrl: "https://api.example.com",
  headers: { Authorization: `Bearer ${process.env.STREAMTIME_TOKEN}` },
});
const org = "<org>";

package main

import (
    "bytes"
    "crypto/rand"
    "encoding/hex"
    "encoding/json"
    "net/http"
    "os"
)

const api = "https://api.example.com/organizations/<org>"

// newKey returns a random idempotency key.
func newKey() string {
    b := make([]byte, 16)
    rand.Read(b)
    return hex.EncodeToString(b)
}

// call sends one request with the access token and a JSON body (or none).
// An optional idempotency key is sent as the Idempotency-Key header.
func call(method, path string, body any, idempotencyKey ...string) (*http.Response, error) {
    var buf bytes.Buffer
    if body != nil {
        if err := json.NewEncoder(&buf).Encode(body); err != nil {
            return nil, err
        }
    }
    req, err := http.NewRequest(method, api+path, &buf)
    if err != nil {
        return nil, err
    }
    req.Header.Set("Authorization", "Bearer "+os.Getenv("STREAMTIME_TOKEN"))
    req.Header.Set("Content-Type", "application/json")
    if len(idempotencyKey) > 0 {
        req.Header.Set("Idempotency-Key", idempotencyKey[0])
    }
    return http.DefaultClient.Do(req)
}

Saving a Virtual Kafka Cluster resource and provisioning it are separate steps in the API: create or change it, then reconcile its new generation. The reconcile returns 202 with the operation it started; the same Idempotency-Key returns that operation again, and a 409 means the generation is out of date.

Create a virtual cluster

UIAPICLIMCP

In the sidebar, select Virtual Kafka, then Add Virtual Kafka Cluster.
Basic Configuration: enter the Identifier and choose the Provider.
Kafka Cluster Selection: choose the Kafka Cluster and set the Capacity Allocation (Kafka Units).
Advanced Configuration: set up client authentication, then create it. It shows Pending, then Provisioning, then Healthy.

HTTPPythonTypeScriptGo

# 1. Save it. The response includes "generation".
curl -X POST "$API/fleet/<fleet>/kafka/kvc/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"identifier": "team-a", "kafka_id": "<kafka-cluster>", "kafka_units": 2}'

# 2. Provision it
curl -X POST "$API/fleet/<fleet>/kafka/kvc/team-a/reconcile/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"generation": 1}'

identifier, kafka_id and kafka_units are required; provider defaults to kroxy. The request is refused when the Kafka cluster has too little capacity left for the Kafka Units you ask for.

import uuid

from streamtime_sdk.api.virtual_kafka_clusters import organizations_fleet_kafka_kvc_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().post(
    f"/organizations/{ORG}/fleet/<fleet>/kafka/kvc/",
    json={"identifier": "team-a", "kafka_id": "<kafka-cluster>", "kafka_units": 2},
)
resp.raise_for_status()

organizations_fleet_kafka_kvc_reconcile_create.sync(
    ORG,
    "<fleet>",
    "team-a",
    client=client,
    body=ReconcileRequest(generation=resp.json()["generation"]),
    idempotency_key=str(uuid.uuid4()),
)

import { organizationsFleetKafkaKvcReconcileCreate } from "./client";

const { data: saved } = await client.post({
  url: "/organizations/{org_identifier}/fleet/{fleet_id}/kafka/kvc/",
  path: { org_identifier: org, fleet_id: "<fleet>" },
  body: { identifier: "team-a", kafka_id: "<kafka-cluster>", kafka_units: 2 },
});

await organizationsFleetKafkaKvcReconcileCreate({
  path: { org_identifier: org, fleet_id: "<fleet>", identifier: "team-a" },
  headers: { "Idempotency-Key": crypto.randomUUID() },
  body: { generation: (saved as { generation: number }).generation },
});

resp, err := call("POST", "/fleet/<fleet>/kafka/kvc/", map[string]any{
    "identifier":  "team-a",
    "kafka_id":    "<kafka-cluster>",
    "kafka_units": 2,
})
if err != nil {
    return err
}
defer resp.Body.Close()

var saved struct {
    Generation int `json:"generation"`
}
if err := json.NewDecoder(resp.Body).Decode(&saved); err != nil {
    return err
}
resp, err = call("POST", "/fleet/<fleet>/kafka/kvc/team-a/reconcile/",
    map[string]any{"generation": saved.Generation}, newKey())

streamtime virtual-kafka-cluster create --config '{
  "identifier": "team-a",
  "kafka_id": "<kafka-cluster>",
  "kafka_units": 2
}'

The CLI saves the virtual cluster but can’t start provisioning it yet. Start it from the UI, or with the API’s reconcile request.

Not available in MCP yet. Create the virtual cluster from the UI, API or CLI.

Check status

status is one of Pending, Provisioning, Updating, Healthy, Error or Deleting.

UIAPICLIMCP

In the sidebar, select Virtual Kafka. Each entry shows its status.
Open it. Overview shows its status, settings and endpoints.

HTTPPythonTypeScriptGo

curl -s "$API/fleet/<fleet>/kafka/kvc/team-a/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

from streamtime_sdk.api.virtual_kafka_clusters import organizations_fleet_kafka_kvc_retrieve

res = organizations_fleet_kafka_kvc_retrieve.sync(ORG, "<fleet>", "team-a", client=client)
print(res.status)

import { organizationsFleetKafkaKvcRetrieve } from "./client";

const { data } = await organizationsFleetKafkaKvcRetrieve({ path: { org_identifier: org, fleet_id: "<fleet>", identifier: "team-a" } });
console.log(data?.status);

resp, err := call("GET", "/fleet/<fleet>/kafka/kvc/team-a/", nil)
// Decode "status" from resp.Body.

streamtime virtual-kafka-cluster list
streamtime virtual-kafka-cluster get team-a

Not available in MCP yet. Check the status from the UI, API or CLI.

Connect clients

Once it’s Healthy, the virtual cluster has its own bootstrap address, a ready-made client configuration and sample commands. Clients sign in with OAuth tokens from the identity provider set in its advanced configuration.

UIAPICLIMCP

Open the virtual cluster from Virtual Kafka.
Endpoints lists its addresses. Select Test for the client properties and sample commands to copy.
Authorization shows the Issuer, Audience, JWKS Endpoint, Authorization URI, Token URI and Scope clients use to get a token.

HTTPPythonTypeScriptGo

# Endpoints, client properties and sample commands are under "resources";
# the OAuth settings under "advanced_config.oauth".
curl -s "$API/fleet/<fleet>/kafka/kvc/team-a/" -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  | jq '{endpoints: .resources.endpoints, client_properties: .resources.client_properties, oauth: .advanced_config.oauth}'

from streamtime_sdk.api.virtual_kafka_clusters import organizations_fleet_kafka_kvc_retrieve

vkc = organizations_fleet_kafka_kvc_retrieve.sync(ORG, "<fleet>", "team-a", client=client)
print(vkc.resources)

import { organizationsFleetKafkaKvcRetrieve } from "./client";

const { data: vkc } = await organizationsFleetKafkaKvcRetrieve({
  path: { org_identifier: org, fleet_id: "<fleet>", identifier: "team-a" },
});
console.log(vkc?.resources);

resp, err := call("GET", "/fleet/<fleet>/kafka/kvc/team-a/", nil)
// Decode "resources" (endpoints, client_properties, commands) from resp.Body.

streamtime virtual-kafka-cluster get team-a --output json

The resources field holds the endpoints and client properties.

Not available in MCP yet. Get the connection details from the UI, API or CLI.

Change the configuration or upgrade

Change the capacity it claims (kafka_units) or its client authentication. Produce and fetch limits follow the new capacity.

UIAPICLIMCP

Open it from Virtual Kafka and select Configuration (or Edit).
Change the settings, then save. Streamtime applies the change, and the status shows Updating, then Healthy.

HTTPPythonTypeScriptGo

# 1. Save the change. The response includes the new "generation".
GEN=$(curl -s -X PATCH "$API/fleet/<fleet>/kafka/kvc/team-a/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"kafka_units": 3}' | jq .generation)

# 2. Apply it
curl -X POST "$API/fleet/<fleet>/kafka/kvc/team-a/reconcile/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d "{\"generation\": $GEN}"

The SDK has no typed call for this update yet, so save it with the SDK’s HTTP client, then apply it:

import uuid

from streamtime_sdk.api.virtual_kafka_clusters import organizations_fleet_kafka_kvc_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().patch(
    f"/organizations/{ORG}/fleet/<fleet>/kafka/kvc/team-a/",
    json={"kafka_units": 3},
)
resp.raise_for_status()

organizations_fleet_kafka_kvc_reconcile_create.sync(
    ORG, "<fleet>", "team-a",
    client=client,
    body=ReconcileRequest(generation=resp.json()["generation"]),
    idempotency_key=str(uuid.uuid4()),
)

import { organizationsFleetKafkaKvcReconcileCreate } from "./client";

const { data: saved } = await client.patch({
  url: "/organizations/{org_identifier}/fleet/<fleet>/kafka/kvc/team-a/",
  path: { org_identifier: org },
  body: {"kafka_units": 3},
});

await organizationsFleetKafkaKvcReconcileCreate({
  path: { org_identifier: org, fleet_id: "<fleet>", identifier: "team-a" },
  headers: { "Idempotency-Key": crypto.randomUUID() },
  body: { generation: (saved as { generation: number }).generation },
});

resp, err := call("PATCH", "/fleet/<fleet>/kafka/kvc/team-a/", json.RawMessage(`{"kafka_units": 3}`))
if err != nil {
    return err
}
defer resp.Body.Close()

var saved struct {
    Generation int `json:"generation"`
}
if err := json.NewDecoder(resp.Body).Decode(&saved); err != nil {
    return err
}
resp, err = call("POST", "/fleet/<fleet>/kafka/kvc/team-a/reconcile/",
    map[string]any{"generation": saved.Generation}, newKey())

Not available in the CLI yet. Change the configuration from the UI or API.

Not available in MCP yet. Change the configuration from the UI or API.

View metrics, logs and alerts

UIAPICLIMCP

Open it from Virtual Kafka and select Monitoring for its metrics and logs, and Alert Rules (where shown) for the alerts that watch it. Alerts that fire also appear on the fleet’s Monitoring tab.

HTTPPythonTypeScriptGo

Metrics and logs are read through the fleet that runs it:

curl -s "$API/fleet/<fleet>/metrics/overview/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -s "$API/fleet/<fleet>/alerts/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

resp = client.get_httpx_client().get(f"/organizations/{ORG}/fleet/<fleet>/alerts/")
print(resp.json())

const { data: alerts } = await client.get({
  url: "/organizations/{org_identifier}/fleet/{fleet_id}/alerts/",
  path: { org_identifier: org, fleet_id: "<fleet>" },
});
console.log(alerts);

resp, err := call("GET", "/fleet/<fleet>/alerts/", nil)

# The fleet's logs from the last hour
streamtime logs fleet <fleet> --since 60

Reading metrics or alerts isn’t available in the CLI yet.

Ask your assistant, for example: “Any alerts on the team-a Virtual Kafka Cluster workload today?” or “Show its error logs from the last hour.”

It uses get_fleet_alerts, get_fleet_logs, list_fleet_metrics and get_fleet_metric on the fleet that runs it.

Delete

Deleting removes what Streamtime installed for it on the fleet.

UIAPICLIMCP

In the sidebar, select Virtual Kafka.
Open its menu in the list, select Delete, and confirm.

The status shows Deleting, then it disappears from the list.

HTTPPythonTypeScriptGo

curl -X DELETE "$API/fleet/<fleet>/kafka/kvc/team-a/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

A draft that was never provisioned is removed at once; otherwise the teardown starts. Poll GET $API/fleet/<fleet>/kafka/kvc/team-a/ until it returns 404.

from streamtime_sdk.api.virtual_kafka_clusters import organizations_fleet_kafka_kvc_destroy

organizations_fleet_kafka_kvc_destroy.sync_detailed(ORG, "<fleet>", "team-a", client=client)

import { organizationsFleetKafkaKvcDestroy } from "./client";

await organizationsFleetKafkaKvcDestroy({ path: { org_identifier: org, fleet_id: "<fleet>", identifier: "team-a" } });

resp, err := call("DELETE", "/fleet/<fleet>/kafka/kvc/team-a/", nil)

streamtime virtual-kafka-cluster delete team-a --yes

Not available in MCP yet. Delete it from the UI, API or CLI.