Kafka

Connect Confluent Cloud or self-hosted Apache Kafka to CloudThinker for stream monitoring, consumer lag, and topic health analysis

Connect your Apache Kafka clusters to enable Alex (Cloud Engineer) and Tony (Database Engineer) to monitor topic health, analyze consumer lag, and optimize streaming performance.

Kafka connections are submitted as a JSON credentials file with separate API key pairs per scope (Confluent Cloud) or a bootstrap address (self-hosted).

Supported platforms

Platform	Support
Confluent Cloud	All tiers
Self-hosted Kafka	2.8+ (KRaft mode), 3.x

Prerequisites

• A Confluent Cloud account with at least one Kafka environment and cluster, OR a self-hosted Kafka 2.8+ (KRaft mode) or 3.x cluster reachable from CloudThinker.
• For Confluent Cloud: admin access to create API keys at confluent.cloud/settings/api-keys.
• Network access from CloudThinker to the cluster’s bootstrap servers and REST endpoints.

Setup

Confluent Cloud

1. Collect the cluster fields

   Go to confluent.cloud, open Environments, and select the environment you want to connect — its ID (env-xxxxx) appears in the URL. Open Clusters, click your target cluster, and collect BOOTSTRAP_SERVERS, KAFKA_REST_ENDPOINT, and KAFKA_CLUSTER_ID. Keep the environment ID as KAFKA_ENV_ID.

2. Create scoped API keys

   Go to Settings → API keys and click + Add API Key. Choose Service Account for production workloads, or My Account for development. Create one key pair per scope you want to enable — Kafka cluster at minimum; Schema Registry, Flink, Cloud resource management, and Tableflow optionally. Save each generated key and secret.

3. Collect optional scope fields

   • Schema Registry: in the environment, open Stream Governance → Schema Registry and collect SCHEMA_REGISTRY_ENDPOINT.
   • Flink: open Flink → Compute pools, click the target pool, and collect FLINK_COMPUTE_POOL_ID; FLINK_ENV_ID is the environment ID and FLINK_REST_ENDPOINT follows your provider and region. FLINK_ORG_ID is under Settings → Organizations.
4. Add the connection in CloudThinker

   Navigate to Connections → Kafka. Create a JSON file with the fields for the scopes you enabled (see the template) and upload it in the connection form.

   Click Connect. CloudThinker verifies the credentials and shows a Connected status.

Scope-based credential model

Each API key and secret pair grants access to one resource scope. Start with Kafka-only fields, then add the other scopes later.

Scope	What it unlocks	Fields
Kafka cluster (required)	Manage topics, produce/consume messages, view cluster metadata	BOOTSTRAP_SERVERS, KAFKA_API_KEY, KAFKA_API_SECRET, KAFKA_CLUSTER_ID, KAFKA_ENV_ID, KAFKA_REST_ENDPOINT
Schema Registry	List, inspect, and delete data schemas	SCHEMA_REGISTRY_ENDPOINT, SCHEMA_REGISTRY_API_KEY, SCHEMA_REGISTRY_API_SECRET
Cloud resource management	Discover environments and clusters, query metrics and billing costs	CONFLUENT_CLOUD_API_KEY, CONFLUENT_CLOUD_API_SECRET
Flink region	Create and manage Flink SQL statements, explore catalogs, run health checks	FLINK_REST_ENDPOINT, FLINK_API_KEY, FLINK_API_SECRET, FLINK_COMPUTE_POOL_ID, FLINK_ENV_ID, FLINK_ORG_ID
Tableflow	Manage Tableflow-enabled topics and catalog integrations	TABLEFLOW_API_KEY, TABLEFLOW_API_SECRET

Connection field template

Fill values for your enabled scopes and remove the blocks you do not use:

{
  "BOOTSTRAP_SERVERS": "pkc-xxxxx.<region>.<provider>.confluent.cloud:9092",
  "KAFKA_API_KEY": "<kafka-api-key>",
  "KAFKA_API_SECRET": "<kafka-api-secret>",
  "KAFKA_REST_ENDPOINT": "https://pkc-xxxxx.<region>.<provider>.confluent.cloud:443",
  "KAFKA_CLUSTER_ID": "lkc-xxxxx",
  "KAFKA_ENV_ID": "env-xxxxx",

  "SCHEMA_REGISTRY_ENDPOINT": "https://psrc-xxxxx.<region>.<provider>.confluent.cloud",
  "SCHEMA_REGISTRY_API_KEY": "<schema-registry-api-key>",
  "SCHEMA_REGISTRY_API_SECRET": "<schema-registry-api-secret>",

  "CONFLUENT_CLOUD_API_KEY": "<cloud-api-key>",
  "CONFLUENT_CLOUD_API_SECRET": "<cloud-api-secret>",

  "FLINK_API_KEY": "<flink-api-key>",
  "FLINK_API_SECRET": "<flink-api-secret>",
  "FLINK_COMPUTE_POOL_ID": "lfcp-xxxxx",
  "FLINK_ENV_ID": "env-xxxxx",
  "FLINK_REST_ENDPOINT": "https://flink.<region>.<provider>.confluent.cloud",
  "FLINK_ORG_ID": "<org-id>",

  "TABLEFLOW_API_KEY": "<tableflow-api-key>",
  "TABLEFLOW_API_SECRET": "<tableflow-api-secret>"
}

Self-hosted Kafka

1. Ensure network access

   Ensure CloudThinker can reach your Kafka broker at its bootstrap address, port 9092 by default.

2. Add the connection in CloudThinker

   Navigate to Connections → Kafka. Create a JSON file with the bootstrap address — and the Schema Registry endpoint, typically port 8081, if you use one — then upload it in the connection form:

   {
     "BOOTSTRAP_SERVERS": "<broker-name>.<your-domain>:9092",
     "SCHEMA_REGISTRY_ENDPOINT": "http://<schema-registry-host>:8081"
   }
   

   Click Connect. CloudThinker verifies the credentials and shows a Connected status.

Connection details

Connection fields are submitted as a JSON credentials file. Fields vary by platform and enabled scope — see the templates in Setup.

Field	Platform	Description
BOOTSTRAP_SERVERS	Both	Kafka cluster bootstrap address (required)
KAFKA_API_KEY / KAFKA_API_SECRET	Confluent Cloud	Kafka scope credentials
SCHEMA_REGISTRY_ENDPOINT	Both	Schema Registry URL
SCHEMA_REGISTRY_API_KEY / SCHEMA_REGISTRY_API_SECRET	Confluent Cloud	Schema Registry credentials
FLINK_REST_ENDPOINT	Confluent Cloud	Flink region endpoint
CONFLUENT_CLOUD_API_KEY / CONFLUENT_CLOUD_API_SECRET	Confluent Cloud	Cloud management credentials
TABLEFLOW_API_KEY / TABLEFLOW_API_SECRET	Confluent Cloud	Tableflow credentials

CloudThinker supports partial scope onboarding — start with Kafka-only fields and add Schema Registry, Flink, Cloud API, or Tableflow credentials later.

Required permissions

Confluent Cloud: create separate API key and secret pairs per scope, from a Service Account. Restrict Kafka ACLs to the specific topics CloudThinker needs. Cloud Management credentials require at minimum the MetricsViewer role.

Self-hosted Kafka: no API keys are required. Ensure the broker’s bootstrap address is network-reachable from CloudThinker on port 9092.

Grant each API key only the scope it needs. Start with Kafka-only credentials and add additional scopes incrementally.

Agent capabilities

Once connected, Alex and Tony can:

Capability	Description
Consumer lag monitoring	Track lag per consumer group, identify slow consumers
Topic health analysis	Check partition distribution, replication factor, under-replicated partitions
Throughput metrics	Monitor bytes in/out, message rates per topic
Broker health	Track broker availability, ISR (In-Sync Replicas) status

Verify the connection

List all Kafka topics and check consumer group lag for the active consumer groups

Example prompts

The slowest consumers behind the lag in the orders-service consumer group
Message throughput trends for the events topic
The topics affected by under-replicated partitions

Troubleshooting

Connection refused or timeout

Verify the broker is running, that the broker port (default 9092) is open and not blocked by a firewall, and that the bootstrap address is correct and reachable from CloudThinker. For local development, ensure Kafka is bound to an accessible IP, not just 127.0.0.1.

Validation errors on a partial-scope JSON file

When using partial scope onboarding, remove the entire key-value pair for unused scopes. Empty-string values cause validation errors:

{
  "BOOTSTRAP_SERVERS": "pkc-xxxxx.<region>.<provider>.confluent.cloud:9092",
  "KAFKA_API_KEY": "<kafka-api-key>",
  "KAFKA_API_SECRET": "<kafka-api-secret>",
  "KAFKA_REST_ENDPOINT": "https://pkc-xxxxx.<region>.<provider>.confluent.cloud:443",
  "KAFKA_CLUSTER_ID": "lkc-xxxxx",
  "KAFKA_ENV_ID": "env-xxxxx"
}

This Kafka-only file is correct because the Schema Registry keys are absent entirely, not present as "".

Schema Registry connection fails

Verify the SCHEMA_REGISTRY_ENDPOINT URL is correct and reachable from CloudThinker. For self-hosted, ensure port 8081 is open. For Confluent Cloud, confirm the Schema Registry API key has the correct permissions for your environment.

Security

• Least privilege — grant only the permissions the agents need for your use case; start read-only and widen later.
• Read-only by default — use read-only credentials unless you want agents to make changes through this connection.
• Rotate credentials — rotate keys and tokens on your normal schedule; CloudThinker picks up the new value when you update the connection.
• Revoke on offboarding — remove the credential at the provider when you delete a connection or a teammate leaves.

• Scope-limited API keys — grant only the scopes CloudThinker needs; start with Kafka-only and add scopes incrementally
• Network restrictions — restrict bootstrap and REST endpoints to CloudThinker’s egress IPs via security groups or firewall rules

Related

Alex Agent

Cloud infrastructure and streaming optimization agent

AWS Connection

Setup instructions for AWS cloud resources