Confluent Platform on VMs

Confluent Platform on VMs installs a stable Confluent Platform release on virtual machines you own, instead of on Kubernetes. Streamtime runs Kafka in KRaft mode and adds Schema Registry, Kafka Connect, REST Proxy and Control Center according to the machines in your VM fleet.

When to use Confluent Platform on VMs

  • No Kubernetes: your organization runs Kafka on virtual machines or bare metal.
  • Existing Confluent Platform estate: you want Streamtime to install, upgrade and monitor Confluent Platform the way you already run it.
  • Dedicated hardware: each cluster gets a fleet of its own.

Prerequisites

  • A VM fleet in Healthy state, with Dedicated tenancy and no other cluster on it. Each VM fleet hosts one Confluent Platform cluster. The fleet’s machines decide which components run where. See Fleet management.
  • A Confluent Platform license, unless you are running a trial.

Cluster settings

Basic Configuration

  • Identifier: unique in your organization, lowercase letters, digits and dashes.
  • Tags: optional key/value labels.

Cloud provider, tenancy and capacity are fixed for this cluster type: it always runs on a VM fleet with Dedicated tenancy, and its size comes from the fleet’s machines, not from Kafka Units.

Fleet Selection (administrators only)

Only VM fleets that are Healthy, Dedicated and empty are listed.

Advanced Configuration (administrators only)

  • Confluent Package Version: the Confluent Platform release to install. The list shows the stable releases you can choose.
  • SASL protocol: scram (SASL/SCRAM-SHA-512, the default) or plain (SASL/PLAIN). Clients always connect over TLS.
  • License: your Confluent Platform license. Leave it blank for a trial.
  • Custom broker properties and custom controller properties: extra Kafka settings as key/value pairs, applied to every broker or controller.

Before you start

  • A fleet in Healthy state with room for the cluster (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:

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)
}

Create a cluster

  1. In the sidebar, select Clusters, then Create cluster.

  2. Cluster Type: choose Confluent Platform (VM), then select Next.

  3. Basic Configuration: enter an Identifier and optional Tags.

  4. Fleet Selection: choose the VM fleet.

  5. Advanced Configuration: choose the Confluent Package Version and the SASL protocol, and add your license.

  6. Select Deploy.

The cluster appears in the list as Pending, then Provisioning, then Healthy.

Find the versions you can install in the confluent_package_version field of the provider’s settings:

curl -s "$API/kafka/configuration-profiles/cpansible/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

Then save the cluster and start provisioning it:

# 1. Save the cluster. The response includes its "generation".
curl -X POST "$API/kafka/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "identifier": "ledger",
    "workload": "cpansible",
    "cloud_provider": "byovm",
    "tenancy_mode": "Dedicated",
    "kafka_units": 1,
    "fleet": "<vm-fleet>",
    "advanced_config": {
      "confluent_package_version": "<version>",
      "sasl_protocol": "scram",
      "license": "<license>"
    }
  }'

# 2. Start provisioning that generation.
curl -X POST "$API/kafka/ledger/reconcile/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"generation": 1}'

fleet and confluent_package_version are required. kafka_units is always 1 for this cluster type. The second request returns 202 with the operation it started; a 409 means the generation is out of date. Poll GET $API/kafka/ledger/ until status is Healthy (or Error).

import uuid

from streamtime_sdk.api.kafka import organizations_kafka_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().post(
    f"/organizations/{ORG}/kafka/",
    json={
        "identifier": "ledger",
        "workload": "cpansible",
        "cloud_provider": "byovm",
        "tenancy_mode": "Dedicated",
        "kafka_units": 1,
        "fleet": "<vm-fleet>",
        "advanced_config": {"confluent_package_version": "<version>"},
    },
)
resp.raise_for_status()

organizations_kafka_reconcile_create.sync(
    ORG,
    "ledger",
    client=client,
    body=ReconcileRequest(generation=resp.json()["generation"]),
    idempotency_key=str(uuid.uuid4()),
)

import { organizationsKafkaReconcileCreate } from "./client";

const { data: created } = await client.post({
  url: "/organizations/{org_identifier}/kafka/",
  path: { org_identifier: org },
  body: {
    identifier: "ledger",
    workload: "cpansible",
    cloud_provider: "byovm",
    tenancy_mode: "Dedicated",
    kafka_units: 1,
    fleet: "<vm-fleet>",
    advanced_config: { confluent_package_version: "<version>" },
  },
});

await organizationsKafkaReconcileCreate({
  path: { org_identifier: org, identifier: "ledger" },
  headers: { "Idempotency-Key": crypto.randomUUID() },
  body: { generation: (created as { generation: number }).generation },
});

resp, err := call("POST", "/kafka/", map[string]any{
    "identifier":      "ledger",
    "workload":        "cpansible",
    "cloud_provider":  "byovm",
    "tenancy_mode":    "Dedicated",
    "kafka_units":     1,
    "fleet":           "<vm-fleet>",
    "advanced_config": map[string]any{"confluent_package_version": "<version>"},
})
if err != nil {
    return err
}
defer resp.Body.Close()

var created struct {
    Generation int `json:"generation"`
}
if err := json.NewDecoder(resp.Body).Decode(&created); err != nil {
    return err
}

// Start provisioning that generation.
resp, err = call("POST", "/kafka/ledger/reconcile/",
    map[string]any{"generation": created.Generation}, newKey())

streamtime kafka create \
  --identifier ledger \
  --workload cpansible \
  --fleet <vm-fleet> \
  --advanced-config '{"confluent_package_version": "<version>"}'

--fleet is required for this cluster type. The CLI sets the VM provider, Dedicated tenancy and one Kafka Unit for you.

The CLI saves the cluster but can’t start provisioning yet. Start it from the API (step 2 in the API tab) or from the cluster’s page in the UI.

Ask your assistant, for example: “Create a Confluent Platform on VMs cluster called ledger on the <vm-fleet> fleet, version <version>.”

It uses create_kafka_cluster with workload cpansible, cloud_provider byovm, tenancy_mode Dedicated and kafka_units 1.

The assistant saves the cluster but can’t start provisioning yet. Start it from the API (step 2 in the API tab) or from the cluster’s page in the UI.

Check status and connection details

  1. In Clusters, find the cluster. The list shows its Status.

  2. Open the cluster. Overview shows the status and the fleet.

  3. To create a user for client applications, select Manage Credentials. The same page gives the connection settings for your clients.
# Status
curl -s "$API/kafka/ledger/status/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Client connection settings
curl -s "$API/kafka/ledger/connection-configuration/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Client credentials: list, then create one
curl -s "$API/kafka/ledger/credentials/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -X POST "$API/kafka/ledger/credentials/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"credential_type": "KAFKA", "username": "ledger-app"}'

This cluster type supports only KAFKA credentials.

from streamtime_sdk.api.kafka_monitoring import organizations_kafka_status_retrieve

status = organizations_kafka_status_retrieve.sync_detailed(ORG, "ledger", client=client)
print(status.content)

conn = client.get_httpx_client().get(
    f"/organizations/{ORG}/kafka/ledger/connection-configuration/"
)
print(conn.json())

import {
  organizationsKafkaConnectionConfigurationRetrieve,
  organizationsKafkaStatusRetrieve,
} from "./client";

const path = { org_identifier: org, kafka_id: "ledger" };
const { data: status } = await organizationsKafkaStatusRetrieve({ path });
const { data: conn } = await organizationsKafkaConnectionConfigurationRetrieve({ path });

resp, err := call("GET", "/kafka/ledger/status/", nil)
resp, err = call("GET", "/kafka/ledger/connection-configuration/", nil)

streamtime kafka get ledger
streamtime kafka status ledger
streamtime kafka credentials ledger

Creating credentials and reading connection settings aren’t available in the CLI yet. Use the UI or API.

Ask, for example: “What is the status of the ledger cluster?” The assistant uses get_cluster_status. MCP doesn’t create credentials.

Change the configuration

A Confluent Platform on VMs cluster takes its size from the VM fleet, so there are no Kafka Units to change. To grow it, add machines to the fleet. You can change the SASL protocol, license and custom broker or controller properties.

Changing these settings isn’t available in the UI yet. Use the API.

# 1. Save the change. The response has the new "generation".
curl -X PATCH "$API/kafka/ledger/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"advanced_config": {
        "confluent_package_version": "<current-version>",
        "custom_broker_properties": {"log.retention.hours": "72"}
      }}'

# 2. Apply that generation.
curl -X POST "$API/kafka/ledger/reconcile/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"generation": <generation>}'

advanced_config is replaced as a whole, so send every setting, including the current confluent_package_version. To change the version, use Upgrade the version instead.

import uuid

from streamtime_sdk.api.kafka import organizations_kafka_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().patch(
    f"/organizations/{ORG}/kafka/ledger/",
    json={"advanced_config": {
        "confluent_package_version": "<current-version>",
        "custom_broker_properties": {"log.retention.hours": "72"},
    }},
)
resp.raise_for_status()

organizations_kafka_reconcile_create.sync(
    ORG,
    "ledger",
    client=client,
    body=ReconcileRequest(generation=resp.json()["generation"]),
    idempotency_key=str(uuid.uuid4()),
)

import {
  organizationsKafkaPartialUpdate,
  organizationsKafkaReconcileCreate,
} from "./client";

const path = { org_identifier: org, identifier: "ledger" };
const { data: cluster } = await organizationsKafkaPartialUpdate({
  path,
  body: {
    advanced_config: {
      confluent_package_version: "<current-version>",
      custom_broker_properties: { "log.retention.hours": "72" },
    },
  },
});
await organizationsKafkaReconcileCreate({
  path,
  headers: { "Idempotency-Key": crypto.randomUUID() },
  body: { generation: cluster!.generation },
});

resp, err := call("PATCH", "/kafka/ledger/", map[string]any{
    "advanced_config": map[string]any{
        "confluent_package_version": "<current-version>",
        "custom_broker_properties":  map[string]string{"log.retention.hours": "72"},
    },
})
if err != nil {
    return err
}
defer resp.Body.Close()

var updated struct {
    Generation int `json:"generation"`
}
if err := json.NewDecoder(resp.Body).Decode(&updated); err != nil {
    return err
}

// Apply that generation.
resp, err = call("POST", "/kafka/ledger/reconcile/",
    map[string]any{"generation": updated.Generation}, newKey())

streamtime kafka update ledger --advanced-config '{"confluent_package_version": "<current-version>", "custom_broker_properties": {"log.retention.hours": "72"}}'

The CLI saves the change but can’t apply it yet. Apply it from the API (step 2 in the API tab).

Not available in MCP yet.

Upgrade the version

Streamtime upgrades through every release between the current and the target version, in order.

  1. Open the cluster and select the Manage tab.
  2. Select Upgrade version and choose the new Version, then confirm.

The cluster shows Upgrading, then Healthy.

# Versions you can upgrade to
curl -s "$API/kafka/ledger/packages/update/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Upgrade to one of them (or send {"strategy": "latest"} for the newest)
curl -X POST "$API/kafka/ledger/packages/update/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"target_version": "<version>"}'

The first request returns current_version and the versions you can choose. Send either target_version or strategy, not both.

http = client.get_httpx_client()
path = f"/organizations/{ORG}/kafka/ledger/packages/update/"

print(http.get(path).json()["versions"])
http.post(path, json={"target_version": "<version>"}).raise_for_status()

const url = "/organizations/{org_identifier}/kafka/{identifier}/packages/update/";
const path = { org_identifier: org, identifier: "ledger" };

const { data: targets } = await client.get({ url, path });
await client.post({ url, path, body: { target_version: "<version>" } });

resp, err := call("GET", "/kafka/ledger/packages/update/", nil)
resp, err = call("POST", "/kafka/ledger/packages/update/",
    map[string]any{"target_version": "<version>"})

Not available in the CLI yet. Use the UI or API.

Not available in MCP yet.

Manage topics

  1. Open the cluster and select Manage Topics.
  2. To add a topic, select Create topic, enter the name, partitions and replication factor, and confirm.
  3. To remove one, select Delete topic and type the topic name to confirm. This deletes the topic’s data.
# List (supports ?search=, ?page=, ?page_size=)
curl -s "$API/kafka/ledger/topics/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Create
curl -X POST "$API/kafka/ledger/topics/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"name": "payments", "partitions": 6, "replication_factor": 3,
       "configs": {"retention.ms": "604800000"}}'

# Add partitions or change settings (partitions can only increase)
curl -X PATCH "$API/kafka/ledger/topics/payments/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"partitions": 12, "configs": {"set": {"retention.ms": "86400000"}}}'

# Delete: repeat the topic name to confirm
curl -X DELETE "$API/kafka/ledger/topics/payments/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"name": "payments"}'

http = client.get_httpx_client()
base = f"/organizations/{ORG}/kafka/ledger/topics/"

http.post(base, json={"name": "payments", "partitions": 6, "replication_factor": 3})
print(http.get(base).json())
http.request("DELETE", base + "payments/", json={"name": "payments"})

import { createKafkaTopic, listKafkaTopics } from "./client";

const path = { org_identifier: org, kafka_id: "ledger" };
await createKafkaTopic({
  path,
  body: { name: "payments", partitions: 6, replication_factor: 3 },
});
const { data } = await listKafkaTopics({ path });

resp, err := call("POST", "/kafka/ledger/topics/", map[string]any{
    "name": "payments", "partitions": 6, "replication_factor": 3,
})
resp, err = call("DELETE", "/kafka/ledger/topics/payments/",
    map[string]any{"name": "payments"})

Not available in the CLI yet. Use the UI or API.

MCP reads topic settings; it doesn’t change them. Ask, for example: “What is the retention of the payments topic on ledger?” (uses get_topic_configuration).

View metrics, logs and alerts

Open the cluster and select the Monitoring tab for throughput, broker health, logs and alerts.

curl -s "$API/kafka/ledger/metrics/overview/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -s "$API/kafka/ledger/logs/"             -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -s "$API/kafka/ledger/firing-alerts/"    -H "Authorization: Bearer $STREAMTIME_TOKEN"

from streamtime_sdk.api.kafka_monitoring import (
    organizations_kafka_firing_alerts_retrieve,
    organizations_kafka_logs_retrieve,
    organizations_kafka_metrics_overview_retrieve,
)

for op in (
    organizations_kafka_metrics_overview_retrieve,
    organizations_kafka_logs_retrieve,
    organizations_kafka_firing_alerts_retrieve,
):
    print(op.sync_detailed(ORG, "ledger", client=client).content)

import {
  organizationsKafkaFiringAlertsRetrieve,
  organizationsKafkaLogsRetrieve,
  organizationsKafkaMetricsOverviewRetrieve,
} from "./client";

const path = { org_identifier: org, kafka_id: "ledger" };
const metrics = await organizationsKafkaMetricsOverviewRetrieve({ path });
const logs = await organizationsKafkaLogsRetrieve({ path });
const alerts = await organizationsKafkaFiringAlertsRetrieve({ path });

for _, p := range []string{"metrics/overview/", "logs/", "firing-alerts/"} {
    resp, err := call("GET", "/kafka/ledger/"+p, nil)
    // ...
}

streamtime kafka metrics ledger
streamtime kafka logs ledger

Alerts aren’t available in the CLI yet.

Ask, for example: “Show me errors in the ledger logs from the last hour” or “Are any alerts firing on ledger?” The assistant uses get_cluster_metric, get_cluster_logs and get_cluster_alerts (list_cluster_metrics lists what can be queried).

Delete the cluster

Deleting a cluster removes its brokers and all topic data. It can’t be undone.

  1. In Clusters, find the cluster and select Delete.
  2. Confirm.

The cluster shows Deleting and then leaves the list.

curl -X DELETE "$API/kafka/ledger/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

Returns 202 with the delete operation; the cluster shows Deleting while its resources are removed, then GET $API/kafka/ledger/ returns 404. A cluster that was never provisioned is removed at once.

from streamtime_sdk.api.kafka import organizations_kafka_destroy

organizations_kafka_destroy.sync_detailed(ORG, "ledger", client=client)

import { organizationsKafkaDestroy } from "./client";

await organizationsKafkaDestroy({
  path: { org_identifier: org, identifier: "ledger" },
});

resp, err := call("DELETE", "/kafka/ledger/", nil)

streamtime kafka delete ledger

Add --yes to skip the confirmation prompt.

Not available in MCP.