Copy page

Kafka

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

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