Apache Kafka

Connect Confluent Cloud or self-hosted Kafka so agents can inspect topics, read messages, check throughput, and debug Flink SQL

Connect a Confluent Cloud or self-hosted Apache Kafka cluster to let CloudThinker agents list topics and their configs, sample messages, query throughput metrics, audit schemas and connectors, and diagnose Flink SQL statements. You connect by uploading one JSON credentials file: a bootstrap address, plus one Confluent API key pair for each scope you want to turn on.

Prerequisites

• Confluent Cloud: an environment with at least one Kafka cluster, and permission to create API keys, ideally for a service account.
• Self-hosted Kafka: a broker whose bootstrap address CloudThinker can reach over the network, port 9092 by default. The file has no settings for SASL/SCRAM, mTLS, client certificates, or a custom CA, so the broker must accept CloudThinker’s connection without them. Don’t open such a broker to the internet: restrict it to CloudThinker’s egress IPs. These docs don’t publish CloudThinker’s egress IP list: ask your CloudThinker contact, or contact us. See Reaching private services for other options.
• A text editor to write the JSON file.
• The Admin role in your CloudThinker workspace, which can add and edit connections. See Roles.

Choose your scopes

On Confluent Cloud, each API key pair unlocks one scope. Start with the Kafka cluster scope and add others later by uploading an updated file. On self-hosted Kafka, only BOOTSTRAP_SERVERS and an optional SCHEMA_REGISTRY_ENDPOINT apply, and every other scope is Confluent Cloud only.

Scope	What agents can do with it	Fields
Kafka cluster (required)	List topics, read topic configs, sample and produce messages, create or delete topics	BOOTSTRAP_SERVERS (both platforms). Confluent Cloud only: KAFKA_API_KEY, KAFKA_API_SECRET, KAFKA_REST_ENDPOINT, KAFKA_CLUSTER_ID, KAFKA_ENV_ID
Cloud resource management	List environments and clusters, query metrics such as received bytes, list billing costs, manage connectors and topic tags	CONFLUENT_CLOUD_API_KEY, CONFLUENT_CLOUD_API_SECRET
Schema Registry	List schema subjects and versions, delete a schema	SCHEMA_REGISTRY_ENDPOINT, SCHEMA_REGISTRY_API_KEY, SCHEMA_REGISTRY_API_SECRET
Flink	Run and inspect Flink SQL statements, browse catalogs and tables, run statement health checks	FLINK_API_KEY, FLINK_API_SECRET, FLINK_REST_ENDPOINT, FLINK_COMPUTE_POOL_ID, FLINK_ENV_ID, FLINK_ORG_ID, optional FLINK_ENV_NAME, FLINK_DATABASE_NAME
Tableflow	List and manage Tableflow topics and catalog integrations	TABLEFLOW_API_KEY, TABLEFLOW_API_SECRET

Setup

Confluent Cloud

1. Collect the cluster fields

   In Confluent Cloud, open Environments and select your environment. Its ID (env-...) is your KAFKA_ENV_ID. Open the cluster and copy its bootstrap server, REST endpoint, and cluster ID (lkc-...).

2. Create one API key pair per scope

   Go to Settings → API keys and click Add API key. Choose a Service account for production. Create a Kafka cluster key first, then one key for each optional scope you chose. Save each key and secret as you create it. Pick the resource that matches each scope:

   Scope	Choose in the wizard	Then select
   Kafka cluster	Kafka cluster	The cluster from step 1
   Cloud resource management	Cloud resource management	Nothing more. It covers the organization
   Schema Registry	Schema Registry	The environment from step 1
   Flink	Flink region	The environment, cloud, and region of your compute pool
   Tableflow	Tableflow	Nothing more

3. Collect optional scope fields

   • Schema Registry: in the environment, open Stream Governance → Schema Registry and copy the endpoint.
   • Flink: open Flink → Compute pools, select the pool, and copy its ID (lfcp-...). The REST endpoint follows the pattern https://flink.<region>.<provider>.confluent.cloud, such as https://flink.us-east-1.aws.confluent.cloud. FLINK_ENV_ID is the env-... ID of the environment that holds the pool, usually the same value as KAFKA_ENV_ID. The organization ID is under Settings → Organizations.
   • Optional Flink defaults: FLINK_ENV_NAME is the environment’s display name, used as the default catalog. FLINK_DATABASE_NAME is the Kafka cluster’s display name, used as the default database. Set them so agents can name tables without the full catalog.database.table path.
4. Write the JSON file

   Save a file such as kafka.json. Keep only the scopes you created keys for, and delete the other lines entirely:

   {
     "BOOTSTRAP_SERVERS": "pkc-xxxxx.us-east-1.aws.confluent.cloud:9092",
     "KAFKA_API_KEY": "<kafka-api-key>",
     "KAFKA_API_SECRET": "<kafka-api-secret>",
     "KAFKA_REST_ENDPOINT": "https://pkc-xxxxx.us-east-1.aws.confluent.cloud:443",
     "KAFKA_CLUSTER_ID": "lkc-xxxxx",
     "KAFKA_ENV_ID": "env-xxxxx",
   
     "CONFLUENT_CLOUD_API_KEY": "<cloud-api-key>",
     "CONFLUENT_CLOUD_API_SECRET": "<cloud-api-secret>",
   
     "SCHEMA_REGISTRY_ENDPOINT": "https://psrc-xxxxx.us-east-1.aws.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_REST_ENDPOINT": "https://flink.us-east-1.aws.confluent.cloud",
     "FLINK_COMPUTE_POOL_ID": "lfcp-xxxxx",
     "FLINK_ENV_ID": "env-xxxxx",
     "FLINK_ORG_ID": "<org-id>",
   
     "TABLEFLOW_API_KEY": "<tableflow-api-key>",
     "TABLEFLOW_API_SECRET": "<tableflow-api-secret>"
   }
   
5. Upload the file in CloudThinker

   Open More → Workspace settings → Connections and select Apache Kafka. Drag the file into the upload area, or click choose a JSON file. Click Connect.

   Success state: the connection shows a Connected status. Run the verify prompt below to confirm agents can see your topics.

Self-hosted Kafka

1. Open network access

   Allow CloudThinker to reach your broker’s bootstrap address, port 9092 by default. A broker bound only to 127.0.0.1 isn’t reachable.

2. Write the JSON file

   Save a file such as kafka.json with the bootstrap address, plus your Schema Registry endpoint if you run one (often port 8081):

   {
     "BOOTSTRAP_SERVERS": "kafka-1.example.com:9092",
     "SCHEMA_REGISTRY_ENDPOINT": "https://schema-registry.example.com:8081"
   }
   
3. Upload the file in CloudThinker

   Open More → Workspace settings → Connections and select Apache Kafka. Upload the file and click Connect.

   Success state: the connection shows a Connected status.

On self-hosted Kafka, agents work through the broker itself: listing topics and sampling or producing messages, plus schemas when you set a Schema Registry endpoint. Metrics, billing, connectors, Flink, and Tableflow are Confluent Cloud APIs, so they aren’t available.

Connection details

The form has one upload field. The file must be one JSON object with string values, using only the keys listed in Choose your scopes. A file with any other key, or a value that isn’t a string, is rejected.

Field	Required	Description
BOOTSTRAP_SERVERS	Yes	Broker bootstrap address, host:port
KAFKA_API_KEY / KAFKA_API_SECRET	Confluent Cloud	Kafka cluster key pair
KAFKA_REST_ENDPOINT, KAFKA_CLUSTER_ID, KAFKA_ENV_ID	Confluent Cloud	Cluster REST endpoint, cluster ID, and environment ID
SCHEMA_REGISTRY_ENDPOINT	No	Schema Registry URL, on either platform
Other scope keys	No	See Choose your scopes

Required permissions

Confluent Cloud: create keys for a service account, and give that account only what agents need.

• Kafka cluster: restrict ACLs to the topics agents should read. Message sampling reads through a consumer group, so the account also needs READ on consumer groups:

  confluent kafka acl create --allow --service-account sa-xxxxx \
    --operations read,describe,describe-configs --topic orders --cluster lkc-xxxxx
  confluent kafka acl create --allow --service-account sa-xxxxx \
    --operations read --consumer-group "*" --cluster lkc-xxxxx
  

• Cloud resource management: at least MetricsViewer for metrics.

• Schema Registry: DeveloperRead on the subjects agents should list.

• Flink: FlinkDeveloper in the environment that holds the compute pool.

Self-hosted Kafka: the connection needs only the bootstrap address.

Start with the Kafka cluster scope and read-only ACLs. Leave write ACLs off the service account unless you want agents to produce messages or create topics.

Agent capabilities

Capability	What agents can do	Goes through Approvals
Topic inventory	List topics with partition counts, read topic configs, search topics by name or tag	No
Message sampling	Read a bounded number of recent messages, decoding with Schema Registry when configured	No
Metrics and billing	Query Confluent Cloud metrics such as received bytes, and list billing costs	No
Schemas and connectors	List schema subjects and versions, list and read connectors	No
Flink SQL reads	Browse catalogs and tables, read statements and their exceptions, run health checks	No
Flink SQL statements	Run a Flink SQL statement	Yes
Produce and create	Produce a message, create topics	Yes
Changes and deletions	Delete topics, schemas, connectors, or Flink statements; alter a topic config; create a connector; delete tags; change Tableflow	Yes

Calls marked Yes go through your workspace Approvals first.

Every change an agent makes goes through your workspace’s Approvals first. On the default Auto · Rules + AI · Ask me mode, AI judgment can let a safe change run without asking — and an Allow Rule can too, once CloudThinker turns on Rule enforcement for your workspace. Choose Manual to approve every change yourself.

A Flink statement can do more than read. An INSERT INTO writes to a topic, and a continuous statement keeps running on your compute pool, using capacity you pay for, until someone deletes it. Read a statement before you approve it.

Partition counts can’t be changed through a topic config. Use the Confluent Cloud UI, or delete and recreate the topic.

Verify the connection

List my Kafka topics with partition counts and replication factor

Expect every topic in the cluster with its partition count and replication factor. On Confluent Cloud with the cloud scope, the agent can also name the environment and cluster.

Example prompts

Show the last 10 messages on the orders topic, decoded with Schema Registry
Chart received bytes per 5 minutes for my Kafka cluster over the last 6 hours
Why is my Flink statement order-enrichment failing?

Change the file later

To add a scope or rotate a key, write a complete file with every scope you want to keep. Open the connection, click the pencil icon on its row, and in Edit Instance click Replace…. Upload the file and click Save and test. If the test fails, CloudThinker keeps the previous credentials. See Rotate a credential.

Troubleshooting

Connection refused or timeout

The broker isn’t reachable from CloudThinker. Check that the broker is running, that 9092 is open in your firewall, and that BOOTSTRAP_SERVERS is correct. A broker bound only to 127.0.0.1 can’t be reached from outside its host.

Validation errors after uploading a partial file

A scope key is present with an empty value. Delete the whole key and value for any scope you don’t use. "SCHEMA_REGISTRY_ENDPOINT": "" fails, while leaving the key out works.

401 Unauthorized

A key pair is wrong, expired, or used for the wrong scope. Each scope needs its own pair: a Kafka cluster key won’t work for Schema Registry or Flink. Create a new pair for that scope and upload an updated file.

Metrics or billing questions fail

The file has no CONFLUENT_CLOUD_API_KEY pair, or its account lacks MetricsViewer. Add the cloud scope and upload the file again. On a self-hosted cluster, these APIs aren’t available.

Schema Registry fails

Check that SCHEMA_REGISTRY_ENDPOINT is reachable from CloudThinker. For self-hosted, open port 8081. For Confluent Cloud, confirm the Schema Registry key belongs to the same 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.

• Self-hosted Kafka takes no credentials: the read-only advice above applies only to Confluent Cloud keys. A self-hosted broker accepts whatever its own configuration allows, so limit network access as below, and remember that produce and create calls still go through Approvals.
• One key pair per scope: give each scope its own key on a service account, so you can revoke Flink or Tableflow access without touching the Kafka cluster key.
• Restrict network access: limit the bootstrap and REST endpoints to CloudThinker’s traffic with firewall rules, especially for a self-hosted broker with no credentials.
• Delete the local file: the JSON file holds every secret in plain text. Delete it after you upload it, or keep it only in a secrets manager.
• Treat messages as sensitive: sampled messages can carry personal data or tokens your producers wrote. Grant topic READ only where agents should look.

Related

Approvals

Decide which Kafka changes run, wait for a person, or are refused.

RabbitMQ connection

Connect another message broker.