# 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](https://confluent.cloud).
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](../kafka/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)

```
[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`):

```bash
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:

```json
{
  "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](https://docs.confluent.io/cloud/current/clusters/cluster-types.html)
- [Confluent Cloud — Create API keys](https://docs.confluent.io/cloud/current/access-management/authenticate/api-keys/api-keys.html)
- [Confluent Cloud — Service accounts](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/service-accounts/manage-service-accounts.html)
- [Configure a Confluent Cloud client](https://docs.confluent.io/cloud/current/client-apps/config-client.html)
- [Monad — Kafka Messages input](../kafka/kafka-messages)