Kafka - CloudThinker

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 Kafka cluster’s bootstrap servers and REST endpoints.

Setup

Select your Kafka platform for specific connection instructions:

Confluent Cloud
Self-hosted Kafka

Open Confluent Cloud and pick your environment

Go to confluent.cloud/home, then open Environments.Click the environment you want to connect.The environment ID appears in the URL after you select it (for example, env-xxxxx).Example navigation:

Environment list: https://confluent.cloud/environments
Selected environment URL pattern: https://confluent.cloud/environments/<env-id>/overview

Get Kafka cluster fields

Inside the selected environment, open Clusters and click your target cluster (for example, <cluster-name>).Collect:

BOOTSTRAP_SERVERS
KAFKA_REST_ENDPOINT
KAFKA_CLUSTER_ID

Keep KAFKA_ENV_ID as the selected environment ID from Step 1.

Create scoped API keys and secrets

Go to confluent.cloud/settings/api-keys and click + Add API Key.Choose Service Account for production workloads, or My Account for development/testing.Select the desired scope in Confluent onboarding, then save the generated API key and API secret pair.Scopes you may create keys for:

Kafka cluster
Schema Registry
ksqlDB cluster
Flink region
Cloud resource management
Tableflow

Get Schema Registry endpoint (optional)

In the selected environment, open Stream Governance -> Schema Registry.Collect:

SCHEMA_REGISTRY_ENDPOINT

URL pattern example: https://confluent.cloud/environments/<env-id>/stream-governance/schema-registry/overview

Get Flink fields (optional)

In the selected environment, open Flink.Open Compute pools and create a pool with + Add compute pool if needed.Click the target compute pool and collect:

FLINK_COMPUTE_POOL_ID
FLINK_ENV_ID (same environment ID from URL)

URL pattern example: https://confluent.cloud/environments/<env-id>/flink/pools/<compute-pool-id>/overviewSet FLINK_REST_ENDPOINT from your cloud provider and region (AWS, Azure, or GCP; for example <region-code>).

Get organization ID (optional)

Go to confluent.cloud/settings/organizations/edit and collect:

FLINK_ORG_ID

Add connection in CloudThinker

In CloudThinker, navigate to Connections → Kafka.Create a JSON file with the fields for the scopes you enabled (see Connection field template below). Upload this JSON file in the connection form.Required fields depend on your profile — see Profiles for details.Click Connect. CloudThinker verifies the credentials and shows a Connected status.

Scope-based credential model

Confluent Cloud uses scope-based API credentials. Each API key and secret pair grants access to a specific resource scope.You can start with Kafka-only fields, then add Schema Registry, Flink, Cloud API, or Tableflow fields later.

Scope	What it unlocks	Typical fields
Kafka cluster	Manage topics (list, create, delete, configure), 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`
Flink region	Create and manage Flink SQL statements, explore catalogs/databases/tables, health checks and diagnostics	`FLINK_REST_ENDPOINT`, `FLINK_API_KEY`, `FLINK_API_SECRET`, `FLINK_COMPUTE_POOL_ID`, `FLINK_ENV_ID`
Cloud resource management	Discover environments and clusters, query operational metrics and billing costs	`CONFLUENT_CLOUD_API_KEY`, `CONFLUENT_CLOUD_API_SECRET`
Tableflow	Manage Tableflow-enabled topics and catalog integrations (e.g., AWS Glue)	`TABLEFLOW_API_KEY`, `TABLEFLOW_API_SECRET`
Organization metadata	Organization-level context for Flink resource management	`FLINK_ORG_ID`

Profiles

Minimal (Kafka-only)

Required:

BOOTSTRAP_SERVERS
KAFKA_API_KEY
KAFKA_API_SECRET
KAFKA_CLUSTER_ID
KAFKA_ENV_ID

What you can do: Manage topics (list, create, delete, configure), produce and consume messages, view cluster metadata and topic configurations.

Standard (Kafka + Schema Registry + Cloud Management)

Add:

SCHEMA_REGISTRY_ENDPOINT
SCHEMA_REGISTRY_API_KEY
SCHEMA_REGISTRY_API_SECRET
CONFLUENT_CLOUD_API_KEY
CONFLUENT_CLOUD_API_SECRET

What you can do: Everything in Minimal, plus list and inspect data schemas, discover environments and clusters, query operational metrics, and view billing costs.

Advanced (Flink / Tableflow)

Add one or more optional scope groups as needed:

Flink: FLINK_REST_ENDPOINT, FLINK_API_KEY, FLINK_API_SECRET, FLINK_COMPUTE_POOL_ID, FLINK_ENV_ID
Tableflow: TABLEFLOW_API_KEY, TABLEFLOW_API_SECRET

What you can do: Everything in Standard, plus create and manage Flink SQL statements, explore Flink catalogs and databases, run health checks on streaming queries, and manage Tableflow-enabled topics with catalog integrations (e.g., AWS Glue).

Connection field template

Use this template and fill values for your enabled scopes:

{
  "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>",

  "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>",

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

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

Ensure network access

Ensure the CloudThinker application can reach your Kafka broker at <broker-name>.<your-domain>:9092.

Add connection in CloudThinker

In CloudThinker, navigate to Connections → Kafka.Create a JSON file with the fields for the scopes you enabled (see Connection field template below). Upload this JSON file in the connection form.Required fields depend on your profile — see Profiles below for details.Click Connect. CloudThinker verifies the credentials and shows a Connected status.

Scope-based credential model

Self-hosted Kafka uses scope-based configuration. You can start with Kafka-only fields, then add Schema Registry fields later.

Scope	What it unlocks	Typical fields
Kafka cluster	Manage topics (list, create, delete), produce/consume messages	`BOOTSTRAP_SERVERS`
Schema Registry	List, inspect, and delete data schemas	`SCHEMA_REGISTRY_ENDPOINT`

Profiles

Minimal (Kafka-only)

Required:

BOOTSTRAP_SERVERS

What you can do: Manage topics (list, create, delete), produce and consume messages.

Standard (with Schema Registry)

Add:

SCHEMA_REGISTRY_ENDPOINT (typically port 8081)

What you can do: Everything in Minimal, plus list, inspect, and delete data schemas.

Connection field template

Use this template and fill values for your enabled scopes:

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

Connection details

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

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

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

Required permissions

For Confluent Cloud, use a Service Account and grant each API key only the scope it needs. Start with Kafka-only credentials and add additional scopes incrementally.

Confluent Cloud: Create separate API key and secret pairs per scope. 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.

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

@alex list all Kafka topics and check consumer group lag for the active consumer groups

Example prompts

@alex check consumer lag for the orders-service group
@tony analyze message throughput trends for the events topic
@alex identify under-replicated partitions and #report the affected topics

Troubleshooting

Connection refused or timeout

Verify the Kafka broker process is running on <broker-name>.<your-domain>.
Check that the broker port (default 9092) is open and not blocked by firewall.
Verify the bootstrap server address <broker-name>.<your-domain>:9092 is correct and reachable from CloudThinker.
For local development, ensure Kafka is bound to an accessible IP (not just 127.0.0.1).

Confluent Cloud partial scope fields

When using partial scope onboarding, remove the entire key-value pair for unused scopes. Do not leave empty strings.Correct (Kafka-only, Schema Registry removed entirely):

{
  "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"
}

Incorrect (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",
  "SCHEMA_REGISTRY_ENDPOINT": "",
  "SCHEMA_REGISTRY_API_KEY": ""
}

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 Schema Registry, Flink, or Cloud Management scopes incrementally
Network restrictions — restrict bootstrap and REST endpoints to CloudThinker’s egress IPs via security groups or firewall rules

Alex Agent

Cloud infrastructure and streaming optimization agent

https://mintcdn.com/cloudthinker/aLd-ttc-SCW-aFky/images/icons/aws.svg?fit=max&auto=format&n=aLd-ttc-SCW-aFky&q=85&s=45d526a3e9345214c0345f277da2e829

AWS Connection

Setup instructions for AWS cloud resources

​Supported platforms

​Prerequisites

​Setup

​Scope-based credential model

​Profiles

​Minimal (Kafka-only)

​Standard (Kafka + Schema Registry + Cloud Management)

​Advanced (Flink / Tableflow)

​Connection field template

​Scope-based credential model

​Profiles

​Minimal (Kafka-only)

​Standard (with Schema Registry)

​Connection field template

​Connection details

​Required permissions

​Agent capabilities

​Verify the connection

​Example prompts

​Troubleshooting

​Security

​Related

Alex Agent

AWS Connection

Supported platforms

Prerequisites

Setup

Scope-based credential model

Profiles

Minimal (Kafka-only)

Standard (Kafka + Schema Registry + Cloud Management)

Advanced (Flink / Tableflow)

Connection field template

Scope-based credential model

Profiles

Minimal (Kafka-only)

Standard (with Schema Registry)

Connection field template

Connection details

Required permissions

Agent capabilities

Verify the connection

Example prompts

Troubleshooting

Security

Related