Skip to main content

​
Overview

The Camunda Zeebe Sink Connector reads JSON business events from Kafka and publishes each record as a Zeebe message. It sits between a Kafka event stream and Camunda Platform 8 processes: the Connector extracts the message name and correlation key from each event and requests message correlation for BPMN process instances waiting for that message. It can also pass the event’s variables and message time-to-live to Zeebe. The input record value must be parseable according to JSON and JSONPath semantics. By default, the messageName and correlationKey fields map to the Zeebe message name and correlation key, variables maps to process variables, and timeToLive maps to the message TTL. The Kafka record key is not used for message mapping. The Connector can connect to Camunda SaaS or a self-managed Zeebe Gateway; it publishes and correlates messages, but does not start processes, complete tasks, or consume Zeebe feedback.

​
Prerequisites

  • For Camunda SaaS mode, prepare the cluster ID, the region when it is not the default region, and a Client ID and Client Secret with access. For self-managed mode, prepare an accessible Zeebe Gateway address and the corresponding authentication method.
  • Each record value in the Kafka input Topic must be a parseable JSON document from which the configured JSONPath expressions can read messageName and correlationKey; if variables or TTL are used, the document must also provide fields convertible to the corresponding types.
  • The Zeebe process model must contain message catch events whose message names and correlation keys match the input; if a message may arrive before the process instance, design the TTL according to the business waiting window.

​
License

Uses Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka, an accessible Zeebe Gateway, and an input Topic, and confirm network connectivity and the required access permissions. For general steps to create and manage a Connector, see AutoMQ’s Manage Connectors. The following example uses a self-managed Zeebe Gateway, a String Converter for the record key, and a JSON Converter for the record value; the record key is not used for JSONPath message mapping. If you use Camunda SaaS, use the SaaS settings in the Configuration section instead.
Replace <kafka-topic> and <zeebe-gateway-host> with the actual resources. The value of each input record should contain at least messageName and correlationKey; variables and timeToLive are optional fields. The record key can be a string, but it does not replace fields in the JSON value and is not used for JSONPath message mapping. For a self-managed Gateway using plaintext transport, set zeebe.client.security.plaintext to true; Camunda SaaS does not use these Gateway settings and instead uses the zeebe.client.cloud.* settings. After the configuration is applied, the Connector publishes each input record as a Zeebe message, and Zeebe correlates it by message name and correlation key.

​
Configuration

​
Zeebe Gateway

​
zeebe.client.gateway.address

The address of the self-managed Zeebe Gateway.
  • Category: Zeebe Gateway
  • Type: string
  • Default: localhost:26500
  • Importance: High
  • Valid values / Notes: Enter an accessible host:port address. Takes effect only when zeebe.client.cloud.clusterId is not set; Camunda SaaS mode ignores this setting.

​
zeebe.client.requestTimeout

The Zeebe request timeout in milliseconds.
  • Category: Zeebe Gateway
  • Type: long
  • Default: 1000
  • Importance: Low
  • Valid values / Notes: Direct Gateway mode uses this value; provide a non-negative duration in milliseconds accepted by the client. The Camunda SaaS client construction branch does not use this setting.

​
zeebe.client.security.plaintext

Whether to use a plaintext connection to the self-managed Zeebe Gateway.
  • Category: Zeebe Gateway
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / Notes: true or false. Only direct Gateway mode reads this setting; use true only when the Gateway is explicitly configured for plaintext transport. Camunda SaaS mode ignores this setting.

​
Camunda SaaS

​
zeebe.client.cloud.clusterId

The Camunda SaaS cluster ID to connect to.
  • Category: Camunda SaaS
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: Setting a non-null value selects the Camunda SaaS client branch; use it together with zeebe.client.cloud.clientId and zeebe.client.cloud.clientSecret. This version does not perform a non-empty validation.

​
zeebe.client.cloud.region

The region where the Camunda SaaS cluster is located.
  • Category: Camunda SaaS
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: This setting overrides the SDK’s region selection only when both zeebe.client.cloud.clusterId and this setting are provided; direct Gateway mode ignores it. The SDK processes the region value; this Connector does not validate the region.

​
Authentication

​
zeebe.client.cloud.clientId

The Client ID used to connect to Camunda SaaS or in direct OAuth mode.
  • Category: Authentication
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: In Camunda SaaS mode, use it with the cluster ID and Client Secret; in direct OAuth mode, setting it enables the OAuth credentials provider and also requires the Client Secret and token audience.

​
zeebe.client.cloud.clientSecret

The Client Secret used to connect to Camunda SaaS or in direct OAuth mode.
  • Category: Authentication
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: Sensitive value. Inject it through controlled configuration and do not write it to shared documents or logs; in Camunda SaaS mode, use it with the cluster ID and Client ID.

​
zeebe.client.cloud.token.audience

The token audience used to request a token in direct OAuth mode.
  • Category: Authentication
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: Used only when zeebe.client.cloud.clusterId is not set and zeebe.client.cloud.clientId is set; the Camunda SaaS client branch ignores it. This Connector does not validate the audience format.

​
zeebe.client.cloud.authorization.server.url

A configuration property representing the authorization server address in the registry.
  • Category: Authentication
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: In this version, the client construction logic does not read this setting, so setting it does not configure a custom token endpoint. Do not rely on it when a custom OAuth endpoint is required.

​
Message Mapping

​
message.path.messageName

The JSONPath expression used to extract the Zeebe message name from the input JSON document.
  • Category: Message Mapping
  • Type: string
  • Default: messageName
  • Importance: High
  • Valid values / Notes: Enter an expression that can be compiled by JSONPath and ensure that every record resolves to a value convertible to a message name. This is a JSONPath expression and does not require a fixed field name.

​
message.path.correlationKey

The JSONPath expression used to extract the Zeebe message correlation key from the input JSON document.
  • Category: Message Mapping
  • Type: string
  • Default: correlationKey
  • Importance: High
  • Valid values / Notes: Enter an expression that can be compiled by JSONPath and ensure that every record resolves to a value convertible to a Zeebe correlation key. Design a stable, business-unique correlation key for each process instance.

​
message.path.variables

The JSONPath expression used to extract Zeebe message variables from the input JSON document.
  • Category: Message Mapping
  • Type: string
  • Default: variables
  • Importance: Medium
  • Valid values / Notes: An empty string disables the variables path. If this setting is enabled but the path is missing from a record, the Connector falls back to using the entire JSON document as variables.

​
message.path.timeToLive

The JSONPath expression used to extract the Zeebe message TTL from the input JSON document.
  • Category: Message Mapping
  • Type: string
  • Default: timeToLive
  • Importance: Low
  • Valid values / Notes: An empty string disables the TTL path; the extracted value is read as milliseconds. If the path is missing, no TTL is sent, and message expiration is determined by Zeebe message semantics.

​
Kafka Connect Sink Basics

​
connector.class

The Kafka Connect Sink Connector class to instantiate.
  • Category: Kafka Connect Sink Basics
  • Type: string
  • Default: No ConfigDef default
  • Importance: High
  • Valid values / Notes: Required; use the fixed value io.zeebe.kafka.connect.ZeebeSinkConnector.

​
tasks.max

The maximum number of Sink Tasks requested for creation.
  • Category: Kafka Connect Sink Basics
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / Notes: Must be at least 1. Actual parallelism also depends on input Topic partition assignment and Zeebe client capacity; increasing this value does not provide ordering guarantees across Tasks.

​
tasks.max.enforce

Whether to enforce the tasks.max limit.
  • Category: Kafka Connect Sink Basics
  • Type: boolean
  • Default: true
  • Importance: Low
  • Deprecated: Yes
  • Valid values / Notes: true or false. Kafka Connect has deprecated this setting and a future major version may remove it; new configurations should not rely on false.

​
key.converter

The Kafka record key Converter at the Connector level.
  • Category: Kafka Connect Sink Basics
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: Must be an instantiable Converter class; when unset, inherits the Worker selection. org.apache.kafka.connect.storage.StringConverter is recommended for record keys; the record key undergoes only Kafka Connect type conversion, does not generate Zeebe message fields, and is not used for JSONPath message mapping.

​
value.converter

The Kafka record value Converter at the Connector level.
  • Category: Kafka Connect Sink Basics
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: Must be an instantiable Converter class; when unset, inherits the Worker selection. The converted value is parsed according to JSON semantics; the Connector does not select a Converter automatically.

​
header.converter

The Kafka record header Converter at the Connector level.
  • Category: Kafka Connect Sink Basics
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: Must be an instantiable HeaderConverter class; when unset, inherits the Worker selection. This Connector does not read record headers directly.

​
transforms

The list of single message transformation (SMT) aliases applied before the SinkTask processes records.
  • Category: Kafka Connect Sink Basics
  • Type: list
  • Default: Empty list
  • Importance: Low
  • Valid values / Notes: Use comma-separated unique aliases; each alias must correspond to a defined transformation. SMTs run before JSONPath extraction.

​
predicates

The list of predicate aliases available to conditional SMTs.
  • Category: Kafka Connect Sink Basics
  • Type: list
  • Default: Empty list
  • Importance: Low
  • Valid values / Notes: Use comma-separated unique aliases; no predicates are enabled by default.

​
config.action.reload

The action taken when an external ConfigProvider value changes.
  • Category: Kafka Connect Sink Basics
  • Type: string
  • Default: restart
  • Importance: Low
  • Valid values / Notes: Valid values are none or restart; this controls the action after an external configuration change and is not a Zeebe client property.

​
Error Handling

​
errors.retry.timeout

The maximum retry duration after a Kafka Connect framework operation fails, in milliseconds.
  • Category: Error Handling
  • Type: long
  • Default: 0
  • Importance: Medium
  • Valid values / Notes: 0 means no retry, and -1 means retry indefinitely. This differs from the Zeebe request timeout and the Connector’s own asynchronous retry mechanism.

​
errors.retry.delay.max.ms

The maximum delay between Kafka Connect framework retries, in milliseconds.
  • Category: Error Handling
  • Type: long
  • Default: 60000
  • Importance: Medium
  • Valid values / Notes: Limits framework retry delay; it does not change the backoff parameters for the Connector’s internal Zeebe request retries.

​
errors.tolerance

The Kafka Connect tolerance policy for framework-level errors.
  • Category: Error Handling
  • Type: string
  • Default: none
  • Importance: Medium
  • Valid values / Notes: Valid values are none or all. all only allows Connect to skip problems identified during framework processing; it does not guarantee that Zeebe publish failures or all SinkTask exceptions are skipped.

​
errors.log.enable

Whether to record tolerated errors and failed operations.
  • Category: Error Handling
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: true or false. This setting does not independently enable a dead letter queue.

​
errors.log.include.messages

Whether to include the Sink record’s Topic, partition, Offset, and timestamp in error logs.
  • Category: Error Handling
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: true or false. Enabling it increases the amount of record metadata exposed; it does not mean that the record value is written to logs.

​
errors.deadletterqueue.topic.name

The dead letter queue Topic used by the Kafka Connect framework error reporter.
  • Category: Error Handling
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: An empty string means that no DLQ is configured. Once configured, this Topic must not also be consumed by topics or matched by topics.regex; a DLQ does not mean that every Zeebe publish failure can be recovered record by record.

​
errors.deadletterqueue.topic.replication.factor

The replication factor used when creating a missing DLQ Topic.
  • Category: Error Handling
  • Type: short
  • Default: 3
  • Importance: Medium
  • Valid values / Notes: Only short type parsing is performed; the Broker’s Topic creation policy and replica constraints still apply.

​
errors.deadletterqueue.context.headers.enable

Whether to write Kafka Connect error-context headers to DLQ records.
  • Category: Error Handling
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: true or false. Takes effect only when a DLQ destination is configured; written headers use the __connect.errors. prefix.

​
Kafka Topic Subscription

​
topics

The list of Kafka Topics consumed by the Connector.
  • Category: Kafka Topic Subscription
  • Type: list
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: Use comma-separated Topic names; this setting is mutually exclusive with topics.regex, and the configured DLQ Topic must not be included in the consumption range.

​
topics.regex

The regular expression for Kafka Topics consumed by the Connector.
  • Category: Kafka Topic Subscription
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: Use Java Pattern syntax; this setting is mutually exclusive with topics, and the expression must not match the configured DLQ Topic.

​
Best Practices

​
Trigger Process Messages with a Stable Correlation Key

Applicable business scenario: Kafka continuously receives order, payment, or inventory events, while BPMN processes in Zeebe wait for the corresponding messages. Each event must be correlated reliably to the target process instance instead of relying on the Kafka record key.
Key notes: correlationKey should map to a stable, business-unique identifier such as an order ID or process business key, and should be designed consistently with the correlation key used by the BPMN message subscription. Do not treat the Connector’s replay behavior as cross-system exactly-once; downstream business actions should still be idempotent. For SaaS, use the zeebe.client.cloud.* settings in the Configuration section instead.

​
Buffer Early Events with a TTL

Applicable business scenario: A business event may enter Kafka before the process instance waiting for the message is created. Zeebe should retain the message for a limited time so that a short difference in arrival order does not immediately lose the correlation opportunity.
Key notes: timeToLive is read in milliseconds, and the TTL is sent only when the input document provides a value through the configured path. TTL expiration means that Zeebe’s message waiting window has ended; it does not mean that the Kafka record was deleted, the business process completed, or the message entered a DLQ. The TTL should cover the acceptable delay before process-instance creation. For SaaS, use the zeebe.client.cloud.* settings in the Configuration section instead.

​
Pass Only the Variables Required by the Process

Applicable business scenario: A Kafka event contains many business fields, but the process needs only a stable, serializable subset. Those fields should be passed as Zeebe message variables without bringing unrelated data into the process context.
Key notes: Put the fields actually required by the process model in the workflowVariables object, and keep their types and structure stable. If the configured variables path is missing, the Connector falls back to sending the entire JSON document; to avoid this fallback, require the object in the production data contract. Message publication does not provide a process-completion acknowledgement; design a separate Zeebe or Kafka feedback stream when an acknowledgement is required. For SaaS, use the zeebe.client.cloud.* settings in the Configuration section instead.

​
Monitoring

​
What to Monitor

Monitor the health and restart counts of the Kafka Connect Worker, Connector, and Tasks, as well as consumption throughput, processing latency, Offset commit progress, consumer lag, errors, and retries. Also observe Worker JVM heap usage, garbage collection, threads, and CPU. When error tolerance or a DLQ is enabled, monitor tolerated errors, DLQ write volume, and DLQ Topic lag separately; when internal Zeebe retries continue, use Task status, request latency, and Zeebe Gateway availability together to determine whether a downstream failure exists.

​
Import the Grafana Dashboard

Download the AutoMQ Connect Cluster Dashboard, select in Grafana a data source that collects Kafka Connect metrics, and ensure that metric labels include the cluster, Connector, and Task identifiers. Then use Grafana’s Import function to import the JSON dashboard.

​
Limitations

  • There is no cross-system transaction between the Connector and Kafka Offset commits, so Kafka-to-Zeebe exactly-once delivery, duplicate-free publishing, and atomic batches are not guaranteed.
  • The Connector does not provide strict message ordering for the same Kafka partition, correlation key, or process instance; after tasks.max is increased, Tasks do not share ordering coordination.
  • Zeebe publish requests within one Kafka Connect batch execute concurrently. If some requests succeed while others fail, there is no transactional rollback, and a restart or an Offset commit gap may publish records again.
  • The Zeebe client performs asynchronous retries only for some gRPC statuses and has no configurable maximum retry count; JSONPath parsing errors and other non-retryable exceptions do not enter this retry mechanism.
  • errors.tolerance and DLQ settings belong to Kafka Connect framework error handling and do not guarantee that Connector-internal Zeebe publish failures are written to the DLQ one by one or skipped.
  • The Connector reads messages only according to JSON/JSONPath semantics; this should not be taken to mean that arbitrary Connect Schemas, field mappings, or Schema evolution combinations are compatible.
  • After a message TTL expires, Zeebe may discard a message that has not been correlated. TTL expiration is not Kafka record deletion, process completion, or confirmation of business processing.

​
FAQ

​
Why does the Task not publish messages after it starts?

Check that exactly one of topics and topics.regex is configured as the subscription selector, confirm that the input Topic has new records, and check whether each value can be parsed through message.path.messageName and message.path.correlationKey. For a self-managed Zeebe deployment, confirm that the Gateway address and TLS or plaintext setting match the server. For SaaS, confirm that the cluster ID, Client ID, and Client Secret are set and that self-managed Gateway settings are not being used accidentally.

​
Why was the message not correlated to a process instance?

First confirm that the message name read by JSONPath matches the BPMN message subscription name, then confirm that the correlation key matches a waiting process instance. If the event may arrive before the process instance, set message.path.timeToLive and provide a sufficient value in milliseconds. After the TTL expires, the message does not continue waiting and no business acknowledgement is generated automatically.

​
What JSON structure do input records require?

By default, the structure must contain at least messageName and correlationKey; for example, an event object can contain those two fields together with variables and timeToLive. If the fields are located elsewhere, change the corresponding message.path.* settings. The value after Converter processing must still conform to the JSON/JSONPath handling supported by the Connector; the Kafka record key does not replace the correlation key in the JSON document.

​
Why might the same message be published again after a restart?

Kafka Connect advances the record position only after put succeeds and the subsequent Offset commit process completes. If a restart, rebalance, or failure occurs after Zeebe has accepted the message but before the Kafka Offset is committed, the record may be delivered again. The Connector generates the same message ID from the same Topic, partition, and Offset. Zeebe may treat an identical existing publish request as successful through ALREADY_EXISTS, but this is not a cross-system exactly-once guarantee; business processing should still be replayable or idempotent.

​
Why are Zeebe publish failures still missing after a dead letter queue is configured?

The DLQ is a Kafka Connect framework error-handling capability and mainly covers errors in framework stages such as Converter and SMT processing. Connector-internal Zeebe publish failures are handled in the SinkTask, so setting errors.deadletterqueue.topic.name alone does not guarantee one record per failure in the DLQ. Also check Task error logs, Zeebe Gateway status, authentication settings, and the Connector’s retry and recovery strategy.