Copy page

Confluent

Messages

Consumes messages from Confluent Cloud Kafka topics with support for SASL authentication (API key/secret), TLS encryption, and consumer group coordination.

Sync Type: Streaming

Details

This input connects directly to a Confluent Cloud Kafka cluster as a Kafka consumer and forwards each message into the pipeline. It uses a consumer group to track offsets, so a Monad pipeline can be paused and resumed without re-reading messages it has already committed.

Prerequisites

Before configuring this input you need:

1. A Confluent Cloud cluster with a topic you want to consume from. You'll need the cluster's bootstrap server endpoint (Console → Cluster → Cluster settings → Endpoints).

2. A Confluent Cloud API Key and Secret scoped to the cluster, with READ permission on the target topic and consumer-group permissions on the consumer group you plan to use. Create one from the Confluent Cloud Console under Cluster → API Keys, or via confluent api-key create --resource <cluster-id>.

3. A consumer group ID — pick a dedicated group ID for Monad (e.g. monad-confluent-consumer) so offset tracking is isolated from any other consumers attached to the same topic.

4. Network reachability between Monad and the Confluent Cloud broker on port 9092 (SASL_SSL).

Setup Instructions

Step 1: Create an API Key in Confluent Cloud

1. Sign in to the Confluent Cloud Console.
2. Open the cluster that holds the topic you want to consume.
3. Go to API Keys → Create key.
4. Pick Service account (recommended) or My account for the key owner.
5. Grant the key READ access on the target topic and the consumer group ID Monad will use.
6. Copy the Key and Secret — the Secret is shown only once.

Step 2: Find the bootstrap server

In the Confluent Cloud Console, open the cluster and go to Cluster settings → Endpoints. Copy the bootstrap server value — it looks like pkc-xxxxx.us-east-1.aws.confluent.cloud:9092.

Step 3: Configure the connector in Monad

1. Bootstrap Servers — paste the endpoint from Step 2.
2. Topic — the topic name on the Confluent cluster.
3. Consumer Group — a dedicated ID for Monad (e.g. monad-confluent-consumer).
4. API Key — the Key from Step 1.
5. API Secret — the Secret from Step 1.
6. Start Offset — earliest to backfill the topic, latest to only forward new messages.
7. Message Format — json if the topic carries JSON payloads, text otherwise.

The connector always authenticates as SASL_SSL / PLAIN — Confluent Cloud's only supported mode — and uses your platform's system trust store for TLS verification.

Configuration

Settings

Setting	Type	Default	Description
Bootstrap Servers	array of strings	—	Confluent Cloud bootstrap servers in host:port format. Typically pkc-xxxxx.region.provider.confluent.cloud:9092.
Topic	string	—	The Kafka topic to consume from. Case-sensitive.
Consumer Group	string	—	Consumer group ID used to coordinate offset tracking. Use a dedicated ID for Monad.
API Key	string	—	Confluent Cloud API Key (used as the SASL username).
API Secret	secret	—	Confluent Cloud API Secret (used as the SASL password).
Start Offset	string	earliest	Where to start when no committed offset exists for the consumer group. Options: earliest, latest.
Message Format	string	json	Expected format of Kafka message values. json expects JSON objects. text wraps any payload in a JSON envelope with Kafka metadata.

The connector authenticates as SASL_SSL / PLAIN unconditionally — Confluent Cloud's only supported mode. No security-protocol selector, no SASL-mechanism selector, no CA Certificate upload. If you need a different Kafka security mode, use the generic Kafka Messages input.

Troubleshooting

Authentication errors

• Verify the API Key is scoped to the cluster you're connecting to. Cluster-scoped keys do not work against other clusters in the same environment.
• Confirm the key has READ permission on the target topic. Without it, the consumer connects but receives no messages and may log authorization errors.

Group Authorization Failed (error code 30)

Code
[30] Group Authorization Failed: the client is not authorized to access a particular group id

In Confluent Cloud, the consumer group ID is its own ACL resource — separate from the topic. Test Connection succeeds because it only validates broker auth + topic metadata, but consumption requires an explicit ACL on the group.

Grant the service account READ on the group (which also covers DESCRIBE):

Code
confluent kafka acl create \
  --allow \
  --service-account <sa-id> \
  --operations read \
  --consumer-group <your-consumer-group-id> \
  --cluster <cluster-id>

Or via the Console: Cluster → ACLs → Add ACL → Service Account → resource type Consumer Group → name = your Consumer Group ID → operation READ.

Use a literal match on the exact Consumer Group ID you configured in Monad (or a prefix match if you prefer, e.g. monad-).

Connection refused

• Confluent Cloud listens on port 9092 for SASL_SSL. Make sure outbound traffic to that port is allowed.
• Bootstrap servers must include the full host:port (port is required, even though :9092 looks redundant).

No messages appearing

• If Start Offset is latest and the consumer group has no committed offset, the consumer only receives messages produced after the pipeline starts. Switch to earliest to read existing messages.
• Confirm the Topic name is spelled exactly as in the Confluent Cloud Console. Topic names are case-sensitive.
• Make sure no other consumer (a separate application or a previous Monad pipeline) is sharing the same Consumer Group ID and pulling messages away.

JSON parse errors

• Switch Message Format to text if the topic carries non-JSON payloads (e.g. Avro, Protobuf, or raw bytes). Schema-Registry-bound formats are not yet supported — use text and decode downstream.

Best Practices

• Dedicated service account + key. Create a Confluent Cloud service account whose API key is used only by Monad. This makes rotation and revocation trivial.
• Grant least-privilege ACLs. The Monad service account needs two ACLs: READ on the topic and READ on the Consumer Group resource (in Confluent Cloud these are separate). Test Connection only checks topic-level auth, so missing the group ACL only surfaces at consume time as Group Authorization Failed.
• Pick a unique Consumer Group ID for each Monad pipeline. Sharing groups across pipelines or with non-Monad consumers causes each consumer to miss messages.
• Rotate API secrets via Monad's secret store, not by editing the configuration in place.

Sample Record

A message ingested with Message Format: json is forwarded as the parsed JSON object. A message ingested with Message Format: text is wrapped in an envelope:

Code
{
  "key": "3f2504e0-4f89-11d3-9a0c-0305e82c3301",
  "value": "event-created",
  "timestamp": "2026-05-21T14:32:10Z",
  "partition": 2,
  "offset": 8471
}

Related Articles

• Confluent Cloud — Cluster settings & endpoints
• Confluent Cloud — Create API keys
• Confluent Cloud — Service accounts
• Configure a Confluent Cloud client
• Monad — Kafka Messages input