Apache Kafka

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-<count>`)	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.

Confluent Cloud
Amazon MSK (IAM)
Self-hosted, Redpanda, Aiven

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:

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).

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

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:

Public cluster
Private cluster, data-plane agent

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. Every advertised broker address must be public too.

Run the data-plane agent 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.”

Get Started

Organizations & Governance

Infrastructure Connectors

Account

CLI

MCP Clients

SDKs & Integrations

Tools

Peek Messages

Safety

Give Danube a read-only principal

Connect

Example prompts

​Tools

​Peek Messages

​Safety

​Give Danube a read-only principal

​Connect

​Example prompts

Tools

Peek Messages

Safety

Give Danube a read-only principal

Connect

Example prompts