> ## Documentation Index > Fetch the complete documentation index at: https://docs.danubeai.com/llms.txt > Use this file to discover all available pages before exploring further. # Apache Kafka > Let agents inspect topics, check consumer group lag and peek at messages in Confluent Cloud, Amazon MSK, Redpanda or your own Kafka cluster The Kafka connector works with any cluster that speaks the Kafka protocol: Confluent Cloud, Amazon MSK (SASL/SCRAM or IAM), Redpanda, Aiven, WarpStream and self-hosted Apache Kafka. It reads cluster metadata, offsets and consumer groups, and it can peek at messages without ever joining a consumer group or committing an offset. ## Tools | Tool | What it returns | | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Test Connection** | Cluster id, controller, every broker's advertised address, whether the principal can describe topics, and warnings about a principal that can write or a listener without TLS. Errors say whether authentication, the network or TLS failed | | **List Topics** | Topics with partition count, replication factor, under-replicated partitions and the internal flag. Filter by name | | **Describe Topic** | Each partition's leader, replicas and in-sync replicas, earliest and latest offsets, an estimate of the retained messages, and the configs set away from the default | | **List Consumer Groups** | Group id, state (`Stable`, `Empty`, `PreparingRebalance`, `Dead`) and protocol type | | **Consumer Group Lag** | Per topic partition: committed offset, log end offset, lag and the consuming member, plus total lag, lag per topic and the group's members | | **Peek Messages** | Messages from a topic: partition, offset, timestamp, key, headers and the value as JSON, text or base64. Avro is decoded through a Schema Registry | | **Produce Message** | Writes one message and returns its partition and offset. Needs a `read_write` connection and a confirmation on every call | ### Peek Messages `from` picks where reading starts: | `from` | Reads | | ----------------------------------------- | ----------------------------------------------------------------------------- | | `latest-50` (default `latest-`) | The newest 50 messages of the topic (or of `partition`) | | `earliest` | From the oldest retained message | | `1200` or `offset:1200` | From offset 1200 in every chosen partition | | `2026-09-26T10:00:00Z` or `1790416800000` | From the first message at or after that time (ISO 8601 or epoch milliseconds) | `count` defaults to 20 and stops at 500. Each value is cut at 16 KB (`value_truncated: true`, with the real `value_size_bytes`). `next_offsets` gives the offset to continue from per partition. When a Schema Registry URL is stored and a value starts with the Confluent wire format (magic byte 0 and a 4-byte schema id), the connector fetches the schema by id. Avro is decoded into JSON. JSON Schema values are parsed as JSON. Protobuf values come back as base64 together with their `schema_id`. Without a registry, every value is returned as JSON, text or base64. ## Safety * **Peek never touches consumer groups.** It assigns partitions directly with no group id and auto-commit off, so it never joins a group, never creates one and never moves a committed offset. Your consumers see nothing. * **Read-only by default.** Every tool except Produce Message only reads. Produce Message only runs on a connection stored with mode `read_write`, and every call returns a `confirm_token` first that the agent must show you and send back. For the strongest guarantee, give the principal Read and Describe permissions only (below): Test Connection warns when it can write. * **Limits on every call.** Calls time out after 15 s by default (up to 55 s with `timeout_seconds`). Lists stop at 500 rows, Peek at 500 messages, and every result at 1 MB. A cut result has `truncated: true`. * **Every broker address is checked.** Kafka clients connect to the brokers' advertised listeners, not to the bootstrap address you stored. Danube checks every bootstrap server before connecting, checks every broker address the cluster advertises before reading anything, and checks each address again when the socket opens, connecting only to the address it approved. A cluster that advertises a private address is refused with `destination_blocked`. * **No secrets in results.** SASL passwords, API secrets, client keys and AWS keys are removed from every error message. MSK IAM tokens are signed per connection from the keys you store; Danube never uses its own AWS credentials. * **Every call is audited.** The audit log records who called, which tool, the duration and the outcome. ## Give Danube a read-only principal The read tools need **Describe** on the cluster, topics and groups and **Read** on topics. Peek Messages does not need any group permission because it never uses a group; Consumer Group Lag needs **Describe** (and on some clusters **Read**) on the group it reports. Create a service account and a **cluster API key** for it (not a Cloud API key). Grant read-only access, either with the `DeveloperRead` role on all topics and consumer groups in the Confluent Cloud console, or with ACLs: ```bash theme={null} confluent kafka acl create --allow --service-account sa-123abc \ --operations read,describe --topic '*' confluent kafka acl create --allow --service-account sa-123abc \ --operations describe,read --consumer-group '*' confluent kafka acl create --allow --service-account sa-123abc \ --operations describe,describe-configs --cluster-scope ``` Store the bootstrap server, security protocol `SASL_SSL`, mechanism `PLAIN`, the API key as the username and the API secret as the password. For Avro, add the Schema Registry endpoint and a Schema Registry API key with `DeveloperRead` on the subjects. Create an IAM user (or role credentials) with this policy, then store the region, access key id and secret access key with mechanism `AWS_MSK_IAM` and security protocol `SASL_SSL`. Use the IAM bootstrap brokers: port 9198 for public access (9098 is the private IAM listener, reached through the data-plane agent). ```json theme={null} { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kafka-cluster:Connect", "kafka-cluster:DescribeCluster", "kafka-cluster:DescribeClusterDynamicConfiguration" ], "Resource": "arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/*" }, { "Effect": "Allow", "Action": [ "kafka-cluster:DescribeTopic", "kafka-cluster:DescribeTopicDynamicConfiguration", "kafka-cluster:ReadData" ], "Resource": "arn:aws:kafka:us-east-1:123456789012:topic/my-cluster/*" }, { "Effect": "Allow", "Action": ["kafka-cluster:DescribeGroup"], "Resource": "arn:aws:kafka:us-east-1:123456789012:group/my-cluster/*" } ] } ``` For MSK with SASL/SCRAM, use mechanism `SCRAM-SHA-512` and the ACLs from the self-hosted tab. Create a SCRAM user (or a client certificate for mTLS) and grant Read and Describe with `kafka-acls`: ```bash theme={null} kafka-acls --bootstrap-server broker:9092 --command-config admin.properties \ --add --allow-principal User:danube \ --operation Read --operation Describe --topic '*' kafka-acls --bootstrap-server broker:9092 --command-config admin.properties \ --add --allow-principal User:danube \ --operation Describe --operation Read --group '*' kafka-acls --bootstrap-server broker:9092 --command-config admin.properties \ --add --allow-principal User:danube \ --operation Describe --operation DescribeConfigs --cluster ``` Redpanda takes the same ACLs through `rpk security acl create`. For mTLS, store the CA, the client certificate and its key, and set the security protocol to `SSL`. ## Connect Open **Apache Kafka** in the dashboard's tool catalog and click **Connect**, or let the agent call `store_credential`. Fill in the bootstrap servers, the security protocol, the SASL mechanism and credentials (or the TLS certificates), and optionally the Schema Registry. Mode stays **Read only** unless you want Produce Message to work. There is no SSH bastion option for Kafka: a tunnel only forwards the bootstrap address, and the client would then follow the advertised broker addresses around it. How Danube reaches the cluster depends on where it lives: Confluent Cloud, MSK with public access, Aiven and Redpanda Cloud connect directly. TLS is on by default (`SASL_SSL`). If the cluster restricts client addresses, allow Danube's egress addresses: see [Connect your production database safely](/connectors/production-database#allowlist-danubes-addresses). Every advertised broker address must be public too. Run the [data-plane agent](/organizations/data-plane) inside the network that can reach the brokers, and store the secrets as references, for example `env://KAFKA_PASSWORD` or `env://MSK_SECRET_ACCESS_KEY`. The call then runs on the agent, which resolves the references locally and applies its `DANUBE_ALLOWED_DESTINATIONS` allowlist to every broker address, including the advertised ones. Nothing inbound is opened, and the secrets never reach Danube. Build the agent image with `DANUBE_DB_ENGINES=kafka`. Run **Test Connection** after saving. It tells you which of these failed: the credentials (`auth_required`), the network (`connection_error`, `destination_blocked`) or TLS (`tls_error`). If Test Connection works but other tools time out, the brokers advertise addresses Danube cannot reach: check the listener's `advertised.listeners`. ## Example prompts * "What's the consumer lag for `payments-worker`, and which partitions are furthest behind?" * "Show me the last 10 messages on `orders` and summarize the statuses." * "Which topics have under-replicated partitions?" * "What's the retention on `audit-events`, and how many messages does it hold?" * "Find the messages on `payments` from 10:00 to 10:05 UTC today with `status: failed`."