Kafka Connector

Source Kafka Docs Node.js Client Docs Connector Changelog

Description

Apache Kafka is an event streaming platform to implement high-performance data pipelines, streaming analytics, data integration, and other applications. This component allows publishing and consuming messages to/from Apache Kafka event streams.

Connections

IAM Access Key and Secret

key: awsAccessKeySecret

An AWS IAM access key pair is required to connect to an Amazon MSK cluster that uses IAM access control. The component signs a SASL/OAUTHBEARER token with the key pair for every broker connection, so no Kafka username or password is involved.

Prerequisites

An Amazon MSK cluster with IAM access control enabled on its brokers
An IAM user or policy that grants the kafka-cluster actions the integration needs (for example kafka-cluster:Connect, kafka-cluster:DescribeTopic, kafka-cluster:ReadData, kafka-cluster:WriteData, kafka-cluster:DescribeGroup, and kafka-cluster:AlterGroup)
Network access from the integration runtime to the cluster's IAM bootstrap brokers

Refer to the MSK client configuration guide for the IAM bootstrap broker endpoints and the permissions each Kafka operation requires.

Setup Steps

To create an IAM access key pair:

Sign in to the AWS Console and navigate to Identity and Access Management (IAM)
Select the IAM user that holds the kafka-cluster permissions, open Security credentials, and select Create access key
Once created, copy the Access key and Secret access key and enter them into the connection configuration of the integration

Configure the Connection

Create a connection of type IAM Access Key and Secret and enter:

Access Key ID: The IAM access key ID
Secret Access Key: The IAM secret access key
AWS Region: The Region of the MSK cluster (e.g., us-east-1), used to sign the authentication token
CA Certificate (optional): A CA certificate in PEM format. This connection always uses TLS. Leave blank to trust the Amazon-issued broker certificates
Enable Avro Deserialization and the Schema Registry fields (optional): Only needed when the Kafka Consumer trigger should decode Avro messages through a Confluent Schema Registry

On every action, trigger, and topic dropdown that uses this connection, also fill in:

Brokers: The cluster's IAM bootstrap broker addresses

Input	Notes	Example
Access Key ID	AWS IAM Access Key ID used for programmatic access. Create access keys in the AWS IAM Console under Security Credentials.	AKIAIOSFODNN7EXAMPLE
Enable Avro Deserialization	Enable Confluent Schema Registry Avro decoding for consumed messages.	false
AWS Region	The AWS Region of the Amazon MSK cluster, used to sign the authentication token.	us-east-1
CA Certificate	Certificate Authority (CA) certificate in PEM format. Leave blank to trust the Amazon-issued broker certificates through the default trust store.	-----BEGIN CERTIFICATE----- MIIDdzCCAl+gAwIBAgIEAgAAuTANBgkqhkiG9w0BAQUFADBaMQswCQYDVQQGEwJJ -----END CERTIFICATE-----
Schema Registry API Key	API key for authenticating with the Schema Registry. Create one in the Confluent Cloud console under the environment's Schema Registry API keys.	ABCDEFGHIJKLMNOP
Schema Registry API Secret	API secret paired with the Schema Registry API key. It is shown once, when the key is created in the Confluent Cloud console.	cfltAbCdEfGhIjKlMnOpQrStUvWxYz0123456789AbCdEfGhIjKlMnOpQrStUvWx
Schema Registry URL	The Confluent Schema Registry endpoint, such as `https://psrc-xxxxx.region.confluent.cloud`, shown in the Confluent Cloud console on the environment's Schema Registry page.	https://psrc-4v5jk.us-east-1.aws.confluent.cloud
Secret Access Key	AWS IAM Secret Access Key paired with the Access Key ID. This value is only shown once when created in the AWS IAM Console.	wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

IAM Role ARN

key: awsAssumeRole

AWS Assume Role authentication connects to an Amazon MSK cluster that uses IAM access control by assuming an IAM role with temporary security credentials. This method is useful for cross account access or when implementing principle of least privilege.

Refer to the AWS documentation on assuming roles for detailed information.

Prerequisites

Completed IAM Access Key and Secret connection setup (access key pair required)
Appropriate permissions to create IAM roles
Network access from the integration runtime to the cluster's IAM bootstrap brokers

Setup Steps

An IAM user with access keys is required to assume a role. If access keys have not been created, follow the IAM Access Key and Secret setup steps first, then return here to create the IAM role.

Create an IAM Role with Trust Policy

Sign in to the AWS Console, open Identity and Access Management (IAM), navigate to Roles and select Create Role
Select Custom trust policy as the trusted entity type
Enter the trust policy below, replacing USER_ARN with the IAM user ARN:

View trust policy template

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "USER_ARN"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

Click Next and attach a policy that grants the kafka-cluster actions the integration needs (for example kafka-cluster:Connect, kafka-cluster:DescribeTopic, kafka-cluster:ReadData, kafka-cluster:WriteData, kafka-cluster:DescribeGroup, and kafka-cluster:AlterGroup)
Complete the remaining steps and select Create Role
Copy the Role ARN from the role summary (format: arn:aws:iam::123456789012:role/role-name)

Configure the Connection

Create a connection of type IAM Role ARN and enter:

Role ARN: The ARN of the IAM role to assume
Access Key ID: From the IAM user (see IAM Access Key and Secret)
Secret Access Key: From the IAM user
External ID (optional): The external ID that the role's trust policy checks with an sts:ExternalId condition, used for cross-account access. Refer to the AWS documentation on external IDs
AWS Region: The Region of the MSK cluster (e.g., us-east-1), used both for the STS call that assumes the role and to sign the authentication token
CA Certificate (optional): A CA certificate in PEM format. Leave blank to trust the Amazon-issued broker certificates
Enable Avro Deserialization and the Schema Registry fields (optional): Only needed when the Kafka Consumer trigger should decode Avro messages through a Confluent Schema Registry

On every action, trigger, and topic dropdown that uses this connection, also fill in:

Brokers: The cluster's IAM bootstrap broker addresses

Verify Connection

The IAM user credentials assume the role, which provides temporary credentials with the role's attached permissions. Ensure the trust policy correctly references the IAM user ARN. The role is assumed once per client and the signed token is reused for every broker connection made by that client.

Input	Notes	Example
Access Key ID	AWS IAM Access Key ID used for programmatic access. Create access keys in the AWS IAM Console under Security Credentials.	AKIAIOSFODNN7EXAMPLE
Enable Avro Deserialization	Enable Confluent Schema Registry Avro decoding for consumed messages.	false
AWS Region	The AWS Region of the Amazon MSK cluster, used to sign the authentication token and to call AWS STS when assuming the role.	us-east-1
CA Certificate	Certificate Authority (CA) certificate in PEM format. Leave blank to trust the Amazon-issued broker certificates through the default trust store.	-----BEGIN CERTIFICATE----- MIIDdzCCAl+gAwIBAgIEAgAAuTANBgkqhkiG9w0BAQUFADBaMQswCQYDVQQGEwJJ -----END CERTIFICATE-----
External ID	A shared value that must match the `sts:ExternalId` condition in the role's trust policy, shown on the role's Trust relationships tab in the IAM console. Optional, but recommended. Please check AWS docs for more information.	shared-common-secret
Role ARN	The ARN of the IAM role to assume, copied from the role's summary page in the IAM console. The role's trust policy must allow the IAM user behind the access key to assume it.	arn:aws:iam::123456789012:role/msk-client
Schema Registry API Key	API key for authenticating with the Schema Registry. Create one in the Confluent Cloud console under the environment's Schema Registry API keys.	ABCDEFGHIJKLMNOP
Schema Registry API Secret	API secret paired with the Schema Registry API key. It is shown once, when the key is created in the Confluent Cloud console.	cfltAbCdEfGhIjKlMnOpQrStUvWxYz0123456789AbCdEfGhIjKlMnOpQrStUvWx
Schema Registry URL	The Confluent Schema Registry endpoint, such as `https://psrc-xxxxx.region.confluent.cloud`, shown in the Confluent Cloud console on the environment's Schema Registry page.	https://psrc-4v5jk.us-east-1.aws.confluent.cloud
Secret Access Key	AWS IAM Secret Access Key paired with the Access Key ID. This value is only shown once when created in the AWS IAM Console.	wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

Basic Authentication

key: basic

Apache Kafka supports authentication using username and password with various SASL mechanisms. This connection requires credentials from the Kafka cluster administrator or from the managed Kafka service provider.

Prerequisites

A Kafka cluster with SASL authentication enabled
Username and password credentials with appropriate permissions
Knowledge of the SASL mechanism configured on the cluster (PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512)
SSL/TLS certificates (if required by the Kafka cluster)

Setup Steps

For managed Kafka services (e.g., Confluent Cloud, Amazon MSK, Azure Event Hubs):

Navigate to the service provider's console
Locate the Kafka cluster and access the security settings
Generate or obtain API credentials (username and API key/password)
Note the SASL mechanism configured for the cluster
Download SSL certificates if required

For self-hosted Kafka clusters:

Contact the Kafka cluster administrator
Request credentials with the appropriate permissions for the use case
Obtain the SASL mechanism type configured on the cluster
Obtain SSL/TLS certificates if the cluster requires secure connections

Refer to the Kafka SASL authentication documentation for more information on authentication mechanisms.

Configure the Connection

Create a connection of type Basic Authentication and enter:

Username: Enter the username provided by the Kafka administrator or service provider
Password: Enter the password or API key
Authentication Mechanism: Select the SASL mechanism configured on the cluster:
- plain - PLAIN SASL mechanism (transmits credentials in plaintext, use with SSL/TLS)
- scram-sha-256 - SCRAM-SHA-256 mechanism (recommended for production)
- scram-sha-512 - SCRAM-SHA-512 mechanism (higher security)
Enable SSL/TLS: Check this option if the Kafka cluster requires secure connections
CA Certificate (if SSL is enabled): Paste the Certificate Authority (CA) certificate in PEM format. This is required for SSL connections. Convert certificates in other formats (JKS, PKCS12) to PEM before pasting.
Client Certificate (if required): Paste the client certificate in PEM format (only if the cluster requires mutual TLS authentication)
Client Key (if required): Paste the client private key in PEM format (only if the cluster requires mutual TLS authentication)
Enable Avro Deserialization and the Schema Registry fields (optional): Only needed when the Kafka Consumer trigger should decode Avro messages through a Confluent Schema Registry

Input	Notes	Example
Authentication Mechanism	Desired authorization method for passing username/password.
Enable Avro Deserialization	Enable Confluent Schema Registry Avro decoding for consumed messages.	false
CA Certificate	Certificate Authority (CA) certificate in PEM format. Leave blank to use the default trust store when the brokers present a publicly trusted certificate.	-----BEGIN CERTIFICATE----- MIIDdzCCAl+gAwIBAgIEAgAAuTANBgkqhkiG9w0BAQUFADBaMQswCQYDVQQGEwJJ -----END CERTIFICATE-----
Client Certificate	Client certificate in PEM format (if required by the Kafka cluster).	-----BEGIN CERTIFICATE----- MIIDdzCCAl+gAwIBAgIEAgAAuTANBgkqhkiG9w0BAQUFADBaMQswCQYDVQQGEwJJ -----END CERTIFICATE-----
Client Key	Client private key in PEM format (if required by the Kafka cluster).	-----BEGIN PRIVATE KEY----- MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC7VJTUt9Us8cKB -----END PRIVATE KEY-----
Password	The SASL password paired with the username, issued alongside it by the cluster administrator or the managed Kafka provider's console.	kafka-password
Schema Registry API Key	API key for authenticating with the Schema Registry. Create one in the Confluent Cloud console under the environment's Schema Registry API keys.	ABCDEFGHIJKLMNOP
Schema Registry API Secret	API secret paired with the Schema Registry API key. It is shown once, when the key is created in the Confluent Cloud console.	cfltAbCdEfGhIjKlMnOpQrStUvWxYz0123456789AbCdEfGhIjKlMnOpQrStUvWx
Schema Registry URL	The Confluent Schema Registry endpoint, such as `https://psrc-xxxxx.region.confluent.cloud`, shown in the Confluent Cloud console on the environment's Schema Registry page.	https://psrc-4v5jk.us-east-1.aws.confluent.cloud
Enable SSL/TLS	Enable SSL/TLS for secure connections.	false
Username	The SASL username used to authenticate with the Kafka brokers, issued by the cluster administrator or the managed Kafka provider's console.	kafka-user

Triggers

Kafka Consumer

Consume messages from Kafka topics on a schedule. | key: kafkaConsumer

Input	Notes	Example
Brokers	A Kafka broker allows consumers to fetch messages by topic, partition and offset.	kafka-broker.example.com:9092
Client ID	The identifier of this Kafka client, sent to the brokers with every request and used in broker logs, metrics, and quotas.	myExampleClient
Connection	The Kafka connection to use.
Consumer Group ID	The unique identifier for the consumer group. Consumers that share this ID split the topic partitions between them and share committed offsets.	my-consumer-group
Consumer Options	Options that control where consumption starts, whether offsets are auto-committed, and whether message keys are deserialized from Avro.
Heartbeat Interval (ms)	How often this consumer sends heartbeats to the group coordinator to keep its session alive, in milliseconds. Must be lower than Session Timeout, typically no more than one third of it.	3000
Max Messages	Maximum number of messages to consume per trigger execution. Must be at least 1. Defaults to 100.	100
Session Timeout (ms)	How long the broker waits without a heartbeat before it removes this consumer from the group and rebalances, in milliseconds.	30000
Topics	List of topics to subscribe to.	my-topic

This trigger consumes messages from Apache Kafka topics on a configured schedule.

Unlike traditional polling triggers that track state between executions, this trigger consumes a configurable number of messages from one or more Kafka topics each time it runs based on the schedule configured in the integration.

How It Works

The trigger executes on the configured schedule (e.g., every 5 minutes)
It connects to Kafka using the provided connection credentials and broker addresses
Creates a consumer with the specified consumer group ID
Subscribes to one or more topics
Consumes up to the maximum number of messages (or times out after 10 seconds if fewer messages are available)
Returns all consumed messages in a structured format
Disconnects from Kafka after consumption

The trigger uses Kafka's consumer group mechanism to track offsets. When Auto Commit is enabled (default), offsets are automatically committed after processing. This ensures that the same messages are not consumed twice by the same consumer group.

Configuration

Configure the following inputs:

Connection: The Kafka connection configuration
Client ID: An identifier for the Kafka consumer passed to the broker with every request (e.g., "myExampleClient")
Brokers: One or more Kafka broker addresses in host:port format (e.g., "kafka-broker.example.com:9092")
Consumer Group ID: The consumer group ID to use for this consumer (e.g., "my-consumer-group")
Topics: One or more topics to subscribe to (e.g., "my-topic")
Max Messages: Maximum number of messages to consume per trigger execution (default: 100)
Session Timeout (ms): The timeout for consumer session in milliseconds (default: 30000)
Heartbeat Interval (ms): The interval for sending heartbeats to the broker in milliseconds (default: 3000)
Consumer Options: A group holding the following fields:
- From Beginning: Whether to start consuming from the beginning of the topic (default: false)
- Auto Commit: Whether to automatically commit offsets after processing messages (default: true)
- Deserialize Keys as Avro: Whether to also deserialize message keys from Avro format. Requires Enable Avro Deserialization on the connection (default: false)

Returned Data

The trigger returns consumed messages along with metadata about the consumption.

Example Payload

{
  "messages": [
    {
      "topic": "my-topic",
      "partition": 0,
      "offset": "42",
      "key": "message-key",
      "value": "{\"orderId\": \"12345\", \"status\": \"shipped\"}",
      "timestamp": "1707500000000",
      "headers": {}
    },
    {
      "topic": "my-topic",
      "partition": 0,
      "offset": "43",
      "key": "message-key-2",
      "value": "{\"orderId\": \"12346\", \"status\": \"delivered\"}",
      "timestamp": "1707500001000",
      "headers": {}
    }
  ],
  "messageCount": 2,
  "consumerGroupId": "my-consumer-group",
  "topics": ["my-topic"]
}

Example Payload for Kafka Consumer⤓

Loading…

Data Sources

Select Topic

Select a Kafka topic from the list. | key: selectTopic | type: picklist

Input	Notes	Example
Brokers	A Kafka broker allows consumers to fetch messages by topic, partition and offset.	kafka-broker.example.com:9092
Client ID	The identifier of this Kafka client, sent to the brokers with every request and used in broker logs, metrics, and quotas.	myExampleClient
Connection	The Kafka connection to use.

Example Payload for Select Topic⤓

Loading…

Actions

Get Consumer Group Status

Get the status and lag information for a consumer group. Specify topics for better performance, or leave empty to check all topics. | key: getConsumerGroupStatus

Input	Notes	Example
Brokers	A Kafka broker allows consumers to fetch messages by topic, partition and offset.	kafka-broker.example.com:9092
Client ID	The identifier of this Kafka client, sent to the brokers with every request and used in broker logs, metrics, and quotas.	myExampleClient
Connection	The Kafka connection to use.
Consumer Group ID	The consumer group ID to check status for.	my-consumer-group
Topics to Check	Specific topics to check for this consumer group. Leave empty to check all topics (slower).	my-topic

Example Payload for Get Consumer Group Status⤓

Loading…

List Topics

List all topics in the Kafka cluster. | key: listTopics

Input	Notes	Example
Brokers	A Kafka broker allows consumers to fetch messages by topic, partition and offset.	kafka-broker.example.com:9092
Client ID	The identifier of this Kafka client, sent to the brokers with every request and used in broker logs, metrics, and quotas.	myExampleClient
Connection	The Kafka connection to use.

Example Payload for List Topics⤓

Loading…

Publish Messages

Publish one or more messages to an Apache Kafka topic. | key: publishMessages

Input	Notes	Example
Brokers	A Kafka broker allows consumers to fetch messages by topic, partition and offset.	kafka-broker.example.com:9092
Client ID	The identifier of this Kafka client, sent to the brokers with every request and used in broker logs, metrics, and quotas.	myExampleClient
Connection	The Kafka connection to use.
Messages	The messages to publish. Each entry's value is sent as the message body; keys are not used.	Hello Kafka
Topic	A Topic is a category/feed name to which records are stored and published.	myTopic

Example Payload for Publish Messages⤓

Loading…

Changelog

2026-10-08

Added Amazon MSK IAM authentication, along with connection and trigger updates:

Added the IAM Access Key and Secret and IAM Role ARN connections for Amazon MSK clusters that use IAM access control
Added a required AWS Region dropdown to the IAM Access Key and Secret and IAM Role ARN connections for selecting the region of the Amazon MSK cluster
Updated the Schema Registry API Key field on the Basic Authentication connection to a masked password field
Updated the Kafka Consumer trigger to group From Beginning, Auto Commit, and Deserialize Keys as Avro into a Consumer Options structured object, and to show Session Timeout (ms) and Heartbeat Interval (ms) as top-level inputs again in place of the Session Timing group
Updated the Kafka Consumer trigger to fail the run when the consumer cannot start, such as when it cannot join its consumer group, and to reject a Max Messages value below 1
Updated the Publish Messages action to close its broker connection when sending messages fails

2026-08-13

Enhanced component capabilities:

Updated the Select Topic data source to read the cluster address from the Brokers input so topic dropdowns resolve (breaking: the data source's Broker input was removed, so previously configured values must be re-entered)
Grouped the Session Timeout and Heartbeat Interval inputs on the Kafka Consumer trigger into a Session Timing structured object
Added output schemas across all actions for improved field mapping during configuration
Added inline action calling support across all actions for improved example output during configuration

2026-07-30

Added Confluent Schema Registry Avro deserialization support for the Kafka Consumer trigger:

Automatically decoded consumed Avro-encoded message values to JSON objects
Added optional Deserialize Keys as Avro input for key deserialization

2026-04-30

Updated spectral version

2026-04-07

Added global debug support across all actions for improved troubleshooting

2026-02-11

Added comprehensive Kafka consumer/subscriber functionality and enhanced connection security.

Kafka Connector

Description​

Connections​

IAM Access Key and Secret​

Prerequisites​

Setup Steps​

Configure the Connection​

IAM Role ARN​

Prerequisites​

Setup Steps​

Create an IAM Role with Trust Policy​

Configure the Connection​

Verify Connection​

Basic Authentication​

Prerequisites​

Setup Steps​

Configure the Connection​

Triggers​

Kafka Consumer​

How It Works​

Configuration​

Returned Data​

Data Sources​

Select Topic​

Actions​

Get Consumer Group Status​

List Topics​

Publish Messages​

Changelog​

2026-10-08​

2026-08-13​

2026-07-30​

2026-04-30​

2026-04-07​

2026-02-11​

Description

Connections

IAM Access Key and Secret

Prerequisites

Setup Steps

Configure the Connection

IAM Role ARN

Prerequisites

Setup Steps

Create an IAM Role with Trust Policy

Configure the Connection

Verify Connection

Basic Authentication

Prerequisites

Setup Steps

Configure the Connection

Triggers

Kafka Consumer

How It Works

Configuration

Returned Data

Data Sources

Select Topic

Actions

Get Consumer Group Status

List Topics

Publish Messages

Changelog

2026-10-08

2026-08-13

2026-07-30

2026-04-30

2026-04-07

2026-02-11