# Messages

Consumes messages from Apache Kafka topics with support for SASL authentication, TLS encryption, and consumer group coordination.

**Sync Type: Streaming**

## Prerequisites

Before configuring this input you need:

1. **A running Kafka cluster** with at least one broker accessible from Monad's network (hostname and port reachable).

2. **A topic** that Monad will consume from.

3. **A consumer group ID** — pick a dedicated group ID for Monad (e.g. `monad-consumer-group`) so offset tracking is isolated from other consumers.

4. **Credentials matching your broker's security protocol** (see the Security Protocol options in [Configuration](#configuration) below):
   - **None** — no credentials required.
   - **SASL\_PLAINTEXT** — a username and password.
   - **SASL + SSL** — a username, password, and your CA certificate in PEM format.
   - **SASL + SSL + mTLS** — a username, password, CA certificate, client certificate, and client private key in PEM format.
   - **SSL** — your CA certificate in PEM format.
   - **SSL + mTLS** — your CA certificate, client certificate, and client private key in PEM format.

## Configuration

### Settings

| Setting | Type |  Default | Description |
|---------|------|---------|-------------|
| Bootstrap Servers | array of strings  | — | List of Kafka broker addresses in `host:port` format. Add multiple brokers for high availability. |
| Topic | string  | — | The Kafka topic to consume messages from. |
| Consumer Group | string  | — | Consumer group ID used to coordinate offset tracking across pipeline runs. Use a dedicated ID for Monad. |
| Security Protocol | one-of  | `NONE` | Security mode for broker connections. See the sub-fields for each protocol below. |
| Start Offset | string  | `earliest` | Where to start consuming when no committed offset exists for the consumer group. Options: `earliest`, `latest`. |
| Message Format | string  | `json` | Expected format of Kafka message values. `json` expects valid JSON objects. `text` wraps any payload in a JSON envelope with Kafka metadata. |

### Security Protocol sub-fields

Each security protocol variant surfaces a different set of credential fields.

**None** — no additional fields.

**SASL Plaintext** — SASL authentication without TLS.

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| SASL Mechanism | string  | Authentication mechanism: `PLAIN`, `SCRAM-SHA-256`, or `SCRAM-SHA-512`. |
| Username | string  | Username for SASL authentication. |
| Password | secret  | Password or API secret for SASL authentication. |

**SASL + SSL** — SASL authentication with TLS encryption.

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| SASL Mechanism | string  | Authentication mechanism: `PLAIN`, `SCRAM-SHA-256`, or `SCRAM-SHA-512`. |
| Username | string  | Username for SASL authentication. |
| Password | secret  | Password or API secret for SASL authentication. |
| CA Certificate | secret  | CA certificate in PEM format used to verify the broker's TLS certificate. |

**SASL + SSL + mTLS** — SASL authentication with TLS encryption and mutual TLS client authentication.

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| SASL Mechanism | string  | Authentication mechanism: `PLAIN`, `SCRAM-SHA-256`, or `SCRAM-SHA-512`. |
| Username | string  | Username for SASL authentication. |
| Password | secret  | Password or API secret for SASL authentication. |
| CA Certificate | secret  | CA certificate in PEM format. |
| Client Certificate | secret  | Client certificate in PEM format for mutual TLS. |
| Client Key | secret  | Client private key in PEM format for mutual TLS. |

**SSL** — TLS encryption without SASL authentication.

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| CA Certificate | secret  | CA certificate in PEM format used to verify the broker's TLS certificate. |

**SSL + mTLS** — TLS encryption with mutual TLS client authentication.

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| CA Certificate | secret  | CA certificate in PEM format. |
| Client Certificate | secret  | Client certificate in PEM format for mutual TLS. |
| Client Key | secret  | Client private key in PEM format for mutual TLS. |

## Troubleshooting

### Connection refused / broker unreachable

- Confirm each broker address in **Bootstrap Servers** is in `host:port` format and reachable from Monad's network.
- Check whether the broker listens on a different port for TLS vs. plaintext (commonly `9092` for plaintext, `9093` for TLS).
- If Monad is hosted in a VPC, ensure the security group or firewall allows outbound connections to the broker ports.

### Authentication errors

- Verify the **Security Protocol** selected in Monad matches the listener configuration on the broker.
- For SASL protocols, confirm the **SASL Mechanism** (`PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`) matches what the broker advertises.
- Ensure the username and password are correct. For Confluent Cloud and MSK the username is often a key ID, not an email address.
- For TLS protocols, check that the CA certificate is the full PEM chain and has not expired.

### TLS / certificate errors

- Paste the CA certificate in full PEM format, including the `-----BEGIN CERTIFICATE-----` and `-----END CERTIFICATE-----` headers.
- For mTLS variants, ensure the client certificate and client private key are a matched pair.
- Self-signed certificates must include the CA that signed the broker certificate, not just the broker certificate itself.

### No messages appearing

- Confirm the **Topic** name is spelled exactly as it appears on the broker (Kafka topic names are case-sensitive).
- If **Start Offset** is `latest` and there is no prior committed offset, Monad will only receive messages produced after the pipeline starts. Change to `earliest` to pick up historical messages.
- Verify the **Consumer Group** ID is not being actively used by another application consuming from the same topic — offset commits from another consumer could advance the position past messages you expect to see.

### Wrong message content / parse errors

- If the pipeline reports JSON parse errors, check that the **Message Format** setting matches the actual payload format produced to the topic.

## Best Practices

- **Use a dedicated consumer group** for Monad (e.g. `monad-consumer-group`). Sharing a group with other applications means those applications and Monad will split the topic's partitions, causing each to miss messages.
- **Prefer `earliest` for new pipelines** when you need full historical data. Switch to `latest` only if you have already captured past messages or only need going-forward data.
- **Use SASL + SSL or SSL** for any topic carrying sensitive data. Plaintext connections transmit credentials and message payloads in the clear.
- **Rotate credentials via Monad secrets management** rather than embedding them directly in the configuration. This makes rotation easier and reduces the blast radius of a leaked secret.
- **Add multiple brokers** to Bootstrap Servers for resilience. If one broker is unavailable, the client can bootstrap from another.

## Sample Record

Messages ingested with **Message Format: json** are forwarded as-is. Messages ingested with **Message Format: text** are wrapped in an envelope:

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