Kafka Connect

A Kafka Connect cluster moves data between a Kafka cluster and other systems (databases, object storage, search) with connectors. Streamtime runs the Connect cluster next to its Kafka cluster, installs the connector plugins you choose, and manages the connectors on it.

Key Features

  • Strimzi or Confluent Connect: choose the Provider to match the Kafka cluster.
  • Your connector plugins: add plugins from HTTPS ZIP archives; the cluster is built with them.
  • Connector lifecycle: browse plugins and their settings, validate a configuration, then create, pause, resume, restart and delete connectors.
  • Monitoring: a dashboard, metrics and logs for the cluster and its connectors.

Settings

Basic Configuration: Identifier and Provider (Strimzi or Confluent).

Kafka Cluster Selection: the Kafka Cluster it connects to.

Advanced Configuration: the connector plugins to install, each a name, an HTTPS URL to a ZIP archive and, optionally, its SHA-512 checksum; and the chart Version, to upgrade.

Before you start

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

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 Kafka Connect 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 Connect cluster

  1. In the sidebar, select Kafka Connect, then Add Kafka Connect.
  2. Basic Configuration: enter the Identifier and choose the Provider.
  3. Kafka Cluster Selection: choose the Kafka Cluster.
  4. Advanced Configuration: add the connector plugins you need.
  5. Create it. It shows Pending, then Provisioning, then Healthy.
# 1. Save it. The response includes "generation".
curl -X POST "$API/kafka-connect/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "identifier": "cdc",
    "kafka_id": "<kafka-cluster>",
    "provider": "skc",
    "advanced_config": {
      "connector_plugin_zips": [{"name": "<plugin>", "url": "https://example.com/<plugin>.zip"}]
    }
  }'

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

provider is skc (Strimzi) or cfk (Confluent). Send "advanced_config": {} to install the default plugins.

import uuid

from streamtime_sdk.api.kafka_connect import organizations_kafka_connect_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().post(
    f"/organizations/{ORG}/kafka-connect/",
    json={
        "identifier": "cdc",
        "kafka_id": "<kafka-cluster>",
        "provider": "skc",
        "advanced_config": {
            "connector_plugin_zips": [{"name": "<plugin>", "url": "https://example.com/<plugin>.zip"}],
        },
    },
)
resp.raise_for_status()

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

import { organizationsKafkaConnectReconcileCreate } from "./client";

const { data: saved } = await client.post({
  url: "/organizations/{org_identifier}/kafka-connect/",
  path: { org_identifier: org },
  body: {
    identifier: "cdc",
    kafka_id: "<kafka-cluster>",
    provider: "skc",
    advanced_config: {
      connector_plugin_zips: [{ name: "<plugin>", url: "https://example.com/<plugin>.zip" }],
    },
  },
});

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

resp, err := call("POST", "/kafka-connect/", map[string]any{
    "identifier": "cdc",
    "kafka_id":   "<kafka-cluster>",
    "provider":   "skc",
    "advanced_config": map[string]any{
        "connector_plugin_zips": []any{
            map[string]any{"name": "<plugin>", "url": "https://example.com/<plugin>.zip"},
        },
    },
})
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", "/kafka-connect/cdc/reconcile/",
    map[string]any{"generation": saved.Generation}, newKey())

streamtime kafka-connect create --config '{
  "identifier": "cdc",
  "kafka_id": "<kafka-cluster>",
  "provider": "skc",
  "advanced_config": {}
}'

The CLI saves the Connect 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 Connect cluster from the UI, API or CLI.

Check status

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

  1. In the sidebar, select Kafka Connect. Each entry shows its status.

  2. Open it. Overview shows its status, settings and endpoints.

curl -s "$API/kafka-connect/cdc/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

from streamtime_sdk.api.kafka_connect import organizations_kafka_connect_retrieve

res = organizations_kafka_connect_retrieve.sync(ORG, "cdc", client=client)
print(res.status)

import { organizationsKafkaConnectRetrieve } from "./client";

const { data } = await organizationsKafkaConnectRetrieve({ path: { org_identifier: org, identifier: "cdc" } });
console.log(data?.status);

resp, err := call("GET", "/kafka-connect/cdc/", nil)
// Decode "status" from resp.Body.

streamtime kafka-connect list
streamtime kafka-connect get cdc

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

Manage connectors

Once the Connect cluster is Healthy, list the plugins it has, look up a plugin’s settings, validate a configuration, then create the connector. A connector’s config is Kafka Connect’s own string key/value map.

  1. Open the Connect cluster from Kafka Connect and select Connectors.
  2. Select Create Connector. Enter the Identifier, choose the Connector type (Source or Sink) and the plugin, and fill in the form generated from the plugin’s settings. Select Test Connection to validate the configuration before you save it.
  3. Open a connector to Pause Connector, Resume Connector, Restart Connector, Update Connector or delete it. Its Status Snapshot shows the state of the connector and its tasks.
C="$API/kafka-connect/cdc"

# Plugins on the cluster, and one plugin's settings
curl -s "$C/connector-plugins/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -s "$C/connector-plugins/<connector-class>/config-schema/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

# Validate, then create
curl -X POST "$C/connectors/validate/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{"connector_class": "<connector-class>", "config": {"tasks.max": "1", "topics": "orders"}}'
curl -X POST "$C/connectors/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" -H "Content-Type: application/json" \
  -d '{
    "identifier": "orders-sink",
    "connector_class": "<connector-class>",
    "plugin_type": "sink",
    "config": {"tasks.max": "1", "topics": "orders"}
  }'

# Status, pause, resume, restart, delete
curl -X POST "$C/connectors/orders-sink/status/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -X POST "$C/connectors/orders-sink/pause/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -X POST "$C/connectors/orders-sink/resume/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -X POST "$C/connectors/orders-sink/restart/" -H "Authorization: Bearer $STREAMTIME_TOKEN"
curl -X DELETE "$C/connectors/orders-sink/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

plugin_type is source or sink. Change a connector’s config with PATCH $C/connectors/orders-sink/.

The typed connector model marks server-set fields as required, so create the connector with the SDK’s HTTP client; the other calls are typed:

from streamtime_sdk.api.kafka_connect import (
    kafka_connector_kafka_connector_pause,
    kafka_connector_status,
    organizations_kafka_connect_connector_plugins_retrieve,
    organizations_kafka_connect_connectors_destroy,
)

plugins = organizations_kafka_connect_connector_plugins_retrieve.sync_detailed(ORG, "cdc", client=client)
print(plugins.content.decode())

http = client.get_httpx_client()
http.post(
    f"/organizations/{ORG}/kafka-connect/cdc/connectors/validate/",
    json={"connector_class": "<connector-class>", "config": {"tasks.max": "1", "topics": "orders"}},
).raise_for_status()
http.post(
    f"/organizations/{ORG}/kafka-connect/cdc/connectors/",
    json={
        "identifier": "orders-sink",
        "connector_class": "<connector-class>",
        "plugin_type": "sink",
        "config": {"tasks.max": "1", "topics": "orders"},
    },
).raise_for_status()

kafka_connector_status.sync_detailed(ORG, "cdc", "orders-sink", client=client)
kafka_connector_kafka_connector_pause.sync_detailed(ORG, "cdc", "orders-sink", client=client)
organizations_kafka_connect_connectors_destroy.sync_detailed(ORG, "cdc", "orders-sink", client=client)

import {
  kafkaConnectorKafkaConnectorPause,
  kafkaConnectorStatus,
  organizationsKafkaConnectConnectorPluginsRetrieve,
  organizationsKafkaConnectConnectorsDestroy,
} from "./client";

const connector = { org_identifier: org, connect_identifier: "cdc", identifier: "orders-sink" };

const { data: plugins } = await organizationsKafkaConnectConnectorPluginsRetrieve({
  path: { org_identifier: org, connect_identifier: "cdc" },
});
console.log(plugins);

await client.post({
  url: "/organizations/{org_identifier}/kafka-connect/{connect_identifier}/connectors/",
  path: { org_identifier: org, connect_identifier: "cdc" },
  body: {
    identifier: "orders-sink",
    connector_class: "<connector-class>",
    plugin_type: "sink",
    config: { "tasks.max": "1", topics: "orders" },
  },
});

await kafkaConnectorStatus({ path: connector });
await kafkaConnectorKafkaConnectorPause({ path: connector });
await organizationsKafkaConnectConnectorsDestroy({ path: connector });

resp, err := call("GET", "/kafka-connect/cdc/connector-plugins/", nil)

resp, err = call("POST", "/kafka-connect/cdc/connectors/", map[string]any{
    "identifier":      "orders-sink",
    "connector_class": "<connector-class>",
    "plugin_type":     "sink",
    "config":          map[string]string{"tasks.max": "1", "topics": "orders"},
})

resp, err = call("POST", "/kafka-connect/cdc/connectors/orders-sink/pause/", nil)
resp, err = call("DELETE", "/kafka-connect/cdc/connectors/orders-sink/", nil)

streamtime kafka-connect connectors plugins list cdc
streamtime kafka-connect connectors schema get cdc <connector-class>

streamtime kafka-connect connectors validate cdc \
  --class <connector-class> \
  --config '{"tasks.max": "1", "topics": "orders"}' --wait

streamtime kafka-connect connectors create cdc --wait --config '{
  "identifier": "orders-sink",
  "connector_class": "<connector-class>",
  "plugin_type": "sink",
  "config": {"tasks.max": "1", "topics": "orders"}
}'

streamtime kafka-connect connectors status cdc orders-sink
streamtime kafka-connect connectors pause cdc orders-sink
streamtime kafka-connect connectors resume cdc orders-sink
streamtime kafka-connect connectors restart cdc orders-sink
streamtime kafka-connect connectors delete cdc orders-sink --yes

Not available in MCP yet. Manage connectors from the UI, API or CLI.

Change the configuration or upgrade

Change the connector plugins, or upgrade to a new Version (advanced_config.chart_config.main.version). The cluster is rebuilt with the new plugins.

  1. Open it from Kafka Connect and select Configuration (or Edit).
  2. Change the settings, then save. Streamtime applies the change, and the status shows Updating, then Healthy.
# 1. Save the change. The response includes the new "generation".
GEN=$(curl -s -X PATCH "$API/kafka-connect/cdc/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"advanced_config": {"connector_plugin_zips": [{"name": "<plugin>", "url": "https://example.com/<plugin>-2.zip"}]}}' | jq .generation)

# 2. Apply it
curl -X POST "$API/kafka-connect/cdc/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.kafka_connect import organizations_kafka_connect_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().patch(
    f"/organizations/{ORG}/kafka-connect/cdc/",
    json={"advanced_config": {"connector_plugin_zips": [{"name": "<plugin>", "url": "https://example.com/<plugin>-2.zip"}]}},
)
resp.raise_for_status()

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

import { organizationsKafkaConnectReconcileCreate } from "./client";

const { data: saved } = await client.patch({
  url: "/organizations/{org_identifier}/kafka-connect/cdc/",
  path: { org_identifier: org },
  body: {"advanced_config": {"connector_plugin_zips": [{"name": "<plugin>", "url": "https://example.com/<plugin>-2.zip"}]}},
});

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

resp, err := call("PATCH", "/kafka-connect/cdc/", json.RawMessage(`{"advanced_config": {"connector_plugin_zips": [{"name": "<plugin>", "url": "https://example.com/<plugin>-2.zip"}]}}`))
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", "/kafka-connect/cdc/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

Open it from Kafka Connect 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.

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 cdc Kafka Connect 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.

  1. In the sidebar, select Kafka Connect.
  2. Open its menu in the list, select Delete, and confirm.

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

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

A draft that was never provisioned is removed at once; otherwise the teardown starts. Poll GET $API/kafka-connect/cdc/ until it returns 404.

from streamtime_sdk.api.kafka_connect import organizations_kafka_connect_destroy

organizations_kafka_connect_destroy.sync_detailed(ORG, "cdc", client=client)

import { organizationsKafkaConnectDestroy } from "./client";

await organizationsKafkaConnectDestroy({ path: { org_identifier: org, identifier: "cdc" } });

resp, err := call("DELETE", "/kafka-connect/cdc/", nil)

streamtime kafka-connect delete cdc --yes

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