> ## Documentation Index > Fetch the complete documentation index at: https://docs.cloudthinker.io/llms.txt > Use this file to discover all available pages before exploring further. # Kafka > Connect Apache Kafka on Confluent Cloud to CloudThinker with scope-based credentials for stream monitoring and management Connect your Apache Kafka clusters to enable [Alex](/guide/agents/alex) (Cloud Engineer) and [Tony](/guide/agents/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: Go to [confluent.cloud/home](https://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//overview` Inside the selected environment, open **Clusters** and click your target cluster (for example, ``). Collect: * `BOOTSTRAP_SERVERS` * `KAFKA_REST_ENDPOINT` * `KAFKA_CLUSTER_ID` Keep `KAFKA_ENV_ID` as the selected environment ID from Step 1. Go to [confluent.cloud/settings/api-keys](https://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 In the selected environment, open **Stream Governance -> Schema Registry**. Collect: * `SCHEMA_REGISTRY_ENDPOINT` URL pattern example: `https://confluent.cloud/environments//stream-governance/schema-registry/overview` 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//flink/pools//overview` Set `FLINK_REST_ENDPOINT` from your cloud provider and region (AWS, Azure, or GCP; for example ``). Go to [confluent.cloud/settings/organizations/edit](https://confluent.cloud/settings/organizations/edit) and collect: * `FLINK_ORG_ID` In CloudThinker, navigate to **Connections → Kafka**. Create a JSON file with the fields for the scopes you enabled (see [Connection field template](#connection-field-template) below). Upload this JSON file in the connection form. Required fields depend on your profile — see [Profiles](#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: ```json theme={null} { "BOOTSTRAP_SERVERS": "pkc-xxxxx...confluent.cloud:9092", "KAFKA_API_KEY": "", "KAFKA_API_SECRET": "", "KAFKA_REST_ENDPOINT": "https://pkc-xxxxx...confluent.cloud:443", "KAFKA_CLUSTER_ID": "lkc-xxxxx", "KAFKA_ENV_ID": "env-xxxxx", "SCHEMA_REGISTRY_ENDPOINT": "https://psrc-xxxxx...confluent.cloud", "SCHEMA_REGISTRY_API_KEY": "", "SCHEMA_REGISTRY_API_SECRET": "", "FLINK_API_KEY": "", "FLINK_API_SECRET": "", "FLINK_COMPUTE_POOL_ID": "lfcp-xxxxx", "FLINK_ENV_ID": "env-xxxxx", "FLINK_REST_ENDPOINT": "https://flink...confluent.cloud", "FLINK_ORG_ID": "", "CONFLUENT_CLOUD_API_KEY": "", "CONFLUENT_CLOUD_API_SECRET": "", "TABLEFLOW_API_KEY": "", "TABLEFLOW_API_SECRET": "" } ``` Ensure the CloudThinker application can reach your Kafka broker at `.:9092`. In CloudThinker, navigate to **Connections → Kafka**. Create a JSON file with the fields for the scopes you enabled (see [Connection field template](#connection-field-template-2) below). Upload this JSON file in the connection form. Required fields depend on your profile — see [Profiles](#profiles-2) 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: ```json theme={null} { "BOOTSTRAP_SERVERS": ".:9092", "SCHEMA_REGISTRY_ENDPOINT": "http://: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](#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](/guide/agents/alex) and [Tony](/guide/agents/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 ```text theme={null} @alex list all Kafka topics and check consumer group lag for the active consumer groups ``` ### Example prompts ```text theme={null} @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 * Verify the Kafka broker process is running on `.`. * Check that the broker port (default 9092) is open and not blocked by firewall. * Verify the bootstrap server address `.:9092` is correct and reachable from CloudThinker. * For local development, ensure Kafka is bound to an accessible IP (not just `127.0.0.1`). 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): ```json theme={null} { "BOOTSTRAP_SERVERS": "pkc-xxxxx...confluent.cloud:9092", "KAFKA_API_KEY": "", "KAFKA_API_SECRET": "", "KAFKA_REST_ENDPOINT": "https://pkc-xxxxx...confluent.cloud:443", "KAFKA_CLUSTER_ID": "lkc-xxxxx", "KAFKA_ENV_ID": "env-xxxxx" } ``` **Incorrect** (empty string values cause validation errors): ```json theme={null} { "BOOTSTRAP_SERVERS": "pkc-xxxxx...confluent.cloud:9092", "KAFKA_API_KEY": "", "KAFKA_API_SECRET": "", "KAFKA_REST_ENDPOINT": "https://pkc-xxxxx...confluent.cloud:443", "KAFKA_CLUSTER_ID": "lkc-xxxxx", "KAFKA_ENV_ID": "env-xxxxx", "SCHEMA_REGISTRY_ENDPOINT": "", "SCHEMA_REGISTRY_API_KEY": "" } ``` 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 *** ## Related Cloud infrastructure and streaming optimization agent Setup instructions for AWS cloud resources