Docs / Strand / connectors/kafka

Apache Kafka Connector

Produce messages to and consume messages from Apache Kafka topics in your Strand workflows.

Prerequisites#

You need Kafka cluster connection details to use this connector. You can connect to a managed Kafka service or a cluster you run yourself.

Your brokers must be reachable from the internet. Strand runs in the cloud

and can only connect to publicly reachable addresses — not 10.x,

172.16–31.x, 192.168.x or 127.x.

A cluster on an internal network needs a public listener or a tunnel; see

Connectors.

Note that the advertised listener matters as much as the bootstrap address: a

broker that advertises a private hostname will fail after the initial connect.

Managed Kafka (Confluent Cloud, AWS MSK, Aiven, etc.)#

1. Sign in to your Kafka provider's dashboard
2. Locate your cluster and find the connection details:

• Bootstrap Servers (e.g., pkc-abc123.us-east-1.aws.confluent.cloud:9092)
• Security Protocol (typically SASL_SSL for managed services)
• SASL Mechanism (typically PLAIN for Confluent Cloud, SCRAM-SHA-256 or SCRAM-SHA-512 for AWS MSK)
• SASL Username (API key or cluster username)
• SASL Password (API secret or cluster password)

Required Permissions#

If your Kafka cluster uses ACLs, the SASL user must have the following permissions:

Tip: In Confluent Cloud, configure ACLs under Cluster settings > ACLs. For self-hosted clusters, use kafka-acls.sh. Scope permissions to specific topic and consumer group names.

Self-Hosted Kafka#

1. Note the Bootstrap Servers of your Kafka cluster (comma-separated list of host:port)
2. Determine the security configuration:

• PLAINTEXT: No authentication or encryption
• SSL: Encryption only
• SASL_PLAINTEXT: Authentication without encryption
• SASL_SSL: Authentication with encryption (recommended)

Tip: Use a dedicated set of credentials for your Strand workflows. For managed services, create an API key scoped to only the topics your workflows need.

Connector Setup#

Create a Kafka connector from the Connectors page.

Configuration Fields#

Operations#

Produce#

Publish a message to a Kafka topic.

Example: Produce an order event:

• Topic: order-events
• Key: {{ payload.customer_id }}
• Message:

{
  "event_type": "order_placed",
  "order_id": "{{ payload.order_id }}",
  "customer_id": "{{ payload.customer_id }}",
  "items": {{ payload.items }},
  "total": {{ payload.total }},
  "timestamp": "{{ payload.timestamp }}"
}

• Headers:

{
  "source": "strand-workflow",
  "correlation-id": "{{ payload.trace_id }}"
}

Example: Produce a notification with a specific partition:

• Topic: notifications
• Partition: 0
• Message: {"type": "alert", "severity": "{{ payload.severity }}", "message": "{{ payload.message }}"}

Consume#

Consume messages from a Kafka topic. Returns a batch of messages up to the specified limit.

A consumer with no committed offset always starts from the earliest available message; this is fixed and not configurable per node.

Example: Consume from an events topic:

• Topic: order-events
• Group ID: strand-order-processor
• Timeout: 5
• Max Messages: 20

Output#

Produce Output#

{
  "success": true,
  "status": "sent",
  "data": {
    "topic": "order-events",
    "partition": 3,
    "offset": 1542,
    "key": "CUST-001",
    "timestamp": "2026-02-18T14:30:00Z"
  },
  "service": "kafka",
  "operation": "produce"
}

Key fields for subsequent nodes:

• {{ steps.node_a1b2c3d4.output_payload.data.topic }}: The topic the message was produced to
• {{ steps.node_a1b2c3d4.output_payload.data.partition }}: The partition the message was written to
• {{ steps.node_a1b2c3d4.output_payload.data.offset }}: The offset of the produced message

Consume Output#

{
  "success": true,
  "status": "completed",
  "data": {
    "messages": [
      {
        "topic": "order-events",
        "partition": 3,
        "offset": 1540,
        "key": "CUST-001",
        "value": "{\"event_type\":\"order_placed\",\"order_id\":\"ORD-001\",\"total\":149.99}",
        "timestamp": "2026-02-18T14:28:00Z",
        "headers": {"source": "strand-workflow"}
      },
      {
        "topic": "order-events",
        "partition": 3,
        "offset": 1541,
        "key": "CUST-002",
        "value": "{\"event_type\":\"order_placed\",\"order_id\":\"ORD-002\",\"total\":89.50}",
        "timestamp": "2026-02-18T14:29:00Z",
        "headers": {"source": "strand-workflow"}
      }
    ],
    "count": 2,
    "group_id": "strand-order-processor"
  },
  "service": "kafka",
  "operation": "consume"
}

Key fields for subsequent nodes:

• {{ steps.node_a1b2c3d4.output_payload.data.messages }}: Array of consumed messages
• {{ steps.node_a1b2c3d4.output_payload.data.count }}: Number of messages consumed
• {{ steps.node_a1b2c3d4.output_payload.data.messages[0].value }}: Value of the first message

Errors#

Example Workflow#

1. Create Connector with your Kafka bootstrap servers and SASL credentials
2. Produce an event when a workflow triggers:

• Topic: workflow-events
• Key: {{ payload.workflow_id }}
• Message: {"event": "started", "workflow_id": "{{ payload.workflow_id }}", "input": {{ payload.input }}}

1. Process the event in downstream workflow nodes
2. Produce a completion event:

• Topic: workflow-events
• Key: {{ payload.workflow_id }}
• Message: {"event": "completed", "workflow_id": "{{ payload.workflow_id }}", "result": "{{ payload.result }}"}
• Headers: {"correlation-id": "{{ payload.trace_id }}"}

Limitations#

• Long-running consumers: The Consume operation performs a one-time poll and returns. It does not maintain a persistent consumer connection. For continuous consumption, use a scheduled workflow or an external consumer that triggers Strand via webhook.
• Exactly-once semantics: Exactly-once delivery is not guaranteed. Messages are produced with at-least-once semantics. Design your consumers to handle duplicates.
• Schema Registry: Avro, Protobuf, and JSON Schema validation via Confluent Schema Registry is not currently supported. Messages are sent and received as plain strings or JSON.
• Message size: Individual messages are limited to the broker's max.message.bytes setting (default: 1 MB).
• Topic management: Creating, deleting, or configuring topics must be done through your Kafka provider or CLI tools.
• Offset management: Consumer offsets are committed automatically after consumption. Manual offset management is not supported.
• Multi-topic: Each operation targets a single topic. Use multiple workflow nodes to interact with different topics.

← Back to Strand docs