Kafka Replicator

A replicator copies topics, and optionally consumer group offsets, from one Kafka cluster to another: between two Streamtime clusters, or between a Streamtime cluster and an external one. Use it for disaster recovery, migrations, or to bring data closer to its readers.

Key Features

• MirrorMaker 2 or Cluster Link, chosen as the Provider.
• Active-passive or active-active replication.
• Choose what to copy: topic and consumer group allow and deny lists (comma-separated names or patterns).
• Offset translation and checkpoints, so consumers can fail over.
• Lag alerts: an alert fires when replication falls behind by a threshold you set, in messages, bytes or seconds.

Settings

• Basic Configuration: Identifier, Provider (MirrorMaker2 or Cluster Link).
• Source Cluster and Destination Cluster: a Streamtime Kafka Cluster, or an External Cluster with its bootstrap servers, security protocol, authentication and the secret holding its credentials.
• Replication Settings: Replication Mode, Topic Whitelist / Topic Blacklist, Groups Whitelist / Groups Blacklist, and the Lag Alert Threshold with its unit.
• Advanced Configuration: MirrorMaker 2 or Cluster Link tuning, such as checkpoints, heartbeats, offset sync topics and producer and consumer settings.

---

Before you start

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

UIAPICLIMCP

1. In the sidebar, select Replicators, then Add Replicator.
2. Basic Configuration: enter the Identifier and choose the Provider.
3. Source Cluster and Destination Cluster: choose a Streamtime Kafka Cluster for each, or enter an External Cluster’s connection details.
4. Replication Settings: choose the Replication Mode, the topics and groups to copy, and the Lag Alert Threshold.
5. Advanced Configuration: tune the replication if you need to, then create it. It shows Pending, then Provisioning, then Healthy.

HTTPPythonTypeScriptGo

# 1. Save it. The response includes "generation".
curl -X POST "$API/kafka-replicator/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "identifier": "orders-dr",
    "provider": "MirrorMaker2",
    "source_kafka_id": "<source-cluster>",
    "destination_kafka_id": "<destination-cluster>",
    "replication_mode": "active-passive",
    "topic_whitelist": "orders.*",
    "lag_alert_threshold": 10000,
    "lag_alert_threshold_unit": "messages"
  }'

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

For an external cluster, send source_connection_details (or destination_connection_details) and the matching …_secret_path instead of the …_kafka_id. replication_mode is active-passive or active-active; lag_alert_threshold_unit is messages, bytes or seconds.

import uuid

from streamtime_sdk.api.kafka_replicator import organizations_kafka_replicator_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().post(
    f"/organizations/{ORG}/kafka-replicator/",
    json={
        "identifier": "orders-dr",
        "provider": "MirrorMaker2",
        "source_kafka_id": "<source-cluster>",
        "destination_kafka_id": "<destination-cluster>",
        "replication_mode": "active-passive",
        "topic_whitelist": "orders.*",
        "lag_alert_threshold": 10000,
        "lag_alert_threshold_unit": "messages",
    },
)
resp.raise_for_status()

organizations_kafka_replicator_reconcile_create.sync(
    ORG,
    "orders-dr",
    client=client,
    body=ReconcileRequest(generation=resp.json()["generation"]),
    idempotency_key=str(uuid.uuid4()),
)

import { organizationsKafkaReplicatorReconcileCreate } from "./client";

const { data: saved } = await client.post({
  url: "/organizations/{org_identifier}/kafka-replicator/",
  path: { org_identifier: org },
  body: {
    identifier: "orders-dr",
    provider: "MirrorMaker2",
    source_kafka_id: "<source-cluster>",
    destination_kafka_id: "<destination-cluster>",
    replication_mode: "active-passive",
    topic_whitelist: "orders.*",
    lag_alert_threshold: 10000,
    lag_alert_threshold_unit: "messages",
  },
});

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

resp, err := call("POST", "/kafka-replicator/", map[string]any{
    "identifier":               "orders-dr",
    "provider":                 "MirrorMaker2",
    "source_kafka_id":          "<source-cluster>",
    "destination_kafka_id":     "<destination-cluster>",
    "replication_mode":         "active-passive",
    "topic_whitelist":          "orders.*",
    "lag_alert_threshold":      10000,
    "lag_alert_threshold_unit": "messages",
})
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-replicator/orders-dr/reconcile/",
    map[string]any{"generation": saved.Generation}, newKey())

streamtime kafka-replicator create --config '{
  "identifier": "orders-dr",
  "provider": "MirrorMaker2",
  "source_kafka_id": "<source-cluster>",
  "destination_kafka_id": "<destination-cluster>",
  "replication_mode": "active-passive",
  "topic_whitelist": "orders.*",
  "lag_alert_threshold": 10000,
  "lag_alert_threshold_unit": "messages"
}'

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

---

Check status

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

UIAPICLIMCP

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

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

HTTPPythonTypeScriptGo

curl -s "$API/kafka-replicator/orders-dr/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

from streamtime_sdk.api.kafka_replicators import organizations_kafka_replicator_retrieve

res = organizations_kafka_replicator_retrieve.sync(ORG, "orders-dr", client=client)
print(res.status)

import { organizationsKafkaReplicatorRetrieve } from "./client";

const { data } = await organizationsKafkaReplicatorRetrieve({ path: { org_identifier: org, identifier: "orders-dr" } });
console.log(data?.status);

resp, err := call("GET", "/kafka-replicator/orders-dr/", nil)
// Decode "status" from resp.Body.

streamtime kafka-replicator list
streamtime kafka-replicator get orders-dr

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

---

Change the configuration or upgrade

Change the replicated topics or groups, the lag threshold, or the advanced replication settings. Replication lag shows on the replicator’s Monitoring tab, and the lag alert fires when it passes the threshold.

UIAPICLIMCP

1. Open it from Replicators and select Configuration (or Edit).
2. 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/kafka-replicator/orders-dr/" \
  -H "Authorization: Bearer $STREAMTIME_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"topic_whitelist": "orders.*,payments.*", "lag_alert_threshold": 5000}' | jq .generation)

# 2. Apply it
curl -X POST "$API/kafka-replicator/orders-dr/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_replicator import organizations_kafka_replicator_reconcile_create
from streamtime_sdk.models import ReconcileRequest

resp = client.get_httpx_client().patch(
    f"/organizations/{ORG}/kafka-replicator/orders-dr/",
    json={"topic_whitelist": "orders.*,payments.*", "lag_alert_threshold": 5000},
)
resp.raise_for_status()

organizations_kafka_replicator_reconcile_create.sync(
    ORG, "orders-dr",
    client=client,
    body=ReconcileRequest(generation=resp.json()["generation"]),
    idempotency_key=str(uuid.uuid4()),
)

import { organizationsKafkaReplicatorReconcileCreate } from "./client";

const { data: saved } = await client.patch({
  url: "/organizations/{org_identifier}/kafka-replicator/orders-dr/",
  path: { org_identifier: org },
  body: {"topic_whitelist": "orders.*,payments.*", "lag_alert_threshold": 5000},
});

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

resp, err := call("PATCH", "/kafka-replicator/orders-dr/", json.RawMessage(`{"topic_whitelist": "orders.*,payments.*", "lag_alert_threshold": 5000}`))
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-replicator/orders-dr/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 Replicators 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 orders-dr Kafka Replicator 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

1. In the sidebar, select Replicators.
2. 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/kafka-replicator/orders-dr/" -H "Authorization: Bearer $STREAMTIME_TOKEN"

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

from streamtime_sdk.api.kafka_replicators import organizations_kafka_replicator_destroy

organizations_kafka_replicator_destroy.sync_detailed(ORG, "orders-dr", client=client)

import { organizationsKafkaReplicatorDestroy } from "./client";

await organizationsKafkaReplicatorDestroy({ path: { org_identifier: org, identifier: "orders-dr" } });

resp, err := call("DELETE", "/kafka-replicator/orders-dr/", nil)

streamtime kafka-replicator delete orders-dr --yes

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