> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Camunda Zeebe Source Connector

> Configure and operate the Camunda Zeebe Source Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting.

## Overview

Camunda Zeebe Source Connector publishes Zeebe Jobs from workflows to Kafka, connecting message-sending steps in a process to event processing, business notifications, and data integration pipelines. It actively claims jobs by Job type, uses a job's custom header to specify the destination Topic, and uses the Job key and the activated job's JSON content as the message key and value.

The Connector also acts as a Zeebe worker and sends job completion commands so that the process can continue. It is not a passive subscriber to process logs and does not export all process events or historical records. Each valid job produces one record for a single Topic. When other workers use the same Job type, the workers compete to claim jobs rather than each receiving a broadcast copy.

## Prerequisites

* The Zeebe process must contain an activatable service task whose Job type matches the Connector's `job.types`. The task's custom header must provide a single valid Kafka Topic name, and the Connector must be allowed to complete the task and advance the process.
* The gRPC gateway for Self-Managed Zeebe typically uses port `26500`; connections must match the gateway's TLS and authentication settings. Camunda SaaS requires a cluster ID, region, and client credentials with the appropriate access permissions.

## License

Uses Apache License 2.0.

## Quick Start

Prepare a Connect Cluster, Kafka, and a local Self-Managed Zeebe 8.4 deployment, and verify access permissions and network connectivity. For Connector creation and management, see [Manage Connectors](../manage-connectors). This example uses a local Zeebe gateway with TLS and authentication disabled. The gateway and Connect Worker share the same network namespace, so the gateway is accessible at `localhost:26500`.

Deploy a process in Zeebe that contains a message-sending service task: set its Job type to `kafka` and its custom header key to `kafka-topic`, with the value `workflow-events`. Prepare this Kafka Topic in advance and start a process instance so that it reaches the service task. Enter the following Connector configuration:

```properties theme={null}
connector.class=io.zeebe.kafka.connect.ZeebeSourceConnector
zeebe.client.security.plaintext=true
key.converter=org.apache.kafka.connect.converters.LongConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
```

The default gateway address is `localhost:26500`, the default Job type is `kafka`, and the default routing header is `kafka-topic`, so the example does not repeat these values. If the Worker runs in a separate container or on a remote host, set `zeebe.client.gateway.address` to a gateway address that the Worker can actually access. Inside a container, `localhost` does not refer to the host or another container. Plaintext connections are appropriate only for the local environment described here, where TLS is explicitly disabled.

After the configuration takes effect, check for messages in `workflow-events` and verify in Zeebe that the task completes and the process continues. The message key is a 64-bit Job key encoded by LongConverter, not a number represented as text. The message value is the activated Job JSON encoded by StringConverter, not just the process variables. Consumers should parse the outer job content before reading variables and should not assume an unconfirmed representation of the variables field. Check Kafka publication and Zeebe job completion separately: completion commands are sent asynchronously, so successful Kafka writes alone do not establish that the process has advanced.

## Configuration

### Connector and Output Encoding

#### `connector.class`

Selects the Zeebe Source Connector implementation.

* **Type**: `string`
* **Default**: No ConfigDef default
* **Importance**: High
* **Required**: Yes
* **Valid Values / Notes**: Use `io.zeebe.kafka.connect.ZeebeSourceConnector` and ensure that the class can be loaded.

#### `tasks.max`

Sets the maximum number of Tasks that Kafka Connect can create.

* **Type**: `int`
* **Default**: `1`
* **Importance**: High
* **Valid Values / Notes**: Must be at least `1`. The Connector assigns types from the `job.types` list; a single distinct type is not automatically split across multiple Tasks. More Tasks than entries in the type list do not provide useful parallelism. Do not use duplicate types as shards.

#### `key.converter`

Overrides the Worker's message key converter.

* **Type**: `class`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Inherits the Worker configuration when unset. The key's Connect type is `INT64`, and its content is the Zeebe Job key. The Quick Start explicitly uses `org.apache.kafka.connect.converters.LongConverter`; this is not the plugin default.

#### `value.converter`

Overrides the Worker's message value converter.

* **Type**: `class`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Inherits the Worker configuration when unset. The value's Connect type is `STRING`, and its content is the activated Job's JSON text. `org.apache.kafka.connect.storage.StringConverter` can encode this text directly. JsonConverter processes a string value; it does not automatically turn it into a structured Connect object.

### Self-Managed Connections and Requests

#### `zeebe.client.gateway.address`

Specifies the Zeebe gateway address for a direct connection.

* **Type**: `string`
* **Default**: `localhost:26500`
* **Importance**: High
* **Valid Values / Notes**: Use the gateway's host and port. Takes effect only when `zeebe.client.cloud.clusterId` is unset; cloud connection mode does not use this setting.

#### `zeebe.client.requestTimeout`

Sets the direct client's default request timeout and the upper bound for Source polling backoff, in milliseconds.

* **Type**: `long`
* **Default**: `1000`
* **Importance**: Low
* **Valid Values / Notes**: ConfigDef declares no range validation; use a usable, nonnegative duration. Activation requests override the client's default request timeout with the current backoff duration, so this setting is not a fixed timeout for every activation request. Cloud mode does not apply it as the client's default request timeout, but Source backoff still reads it.

#### `zeebe.client.security.plaintext`

Controls whether a direct connection disables transport-layer TLS.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Low
* **Valid Values / Notes**: Set to `true` only when explicitly using a plaintext gateway. Cloud mode ignores this setting. Do not disable secure connections to bypass TLS configuration problems.

### Cloud Cluster Selection

#### `zeebe.client.cloud.clusterId`

Selects Camunda SaaS connection mode and the cluster.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Any non-`null` value selects cloud mode, including an empty string. For Self-Managed deployments, remove this setting rather than supplying an empty string. Cloud mode requires the corresponding client credentials. The Connector does not check credential completeness itself and does not use the direct gateway address, plaintext switch, or OAuth audience settings.

#### `zeebe.client.cloud.region`

Specifies the cloud cluster region.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Passed to the client only in cloud mode and when non-`null`. It should match the cluster's actual region. When unset, the SDK handles it; this does not mean that the Connector provides a fixed default region.

### Authentication

#### `zeebe.client.cloud.clientId`

Specifies the OAuth client ID.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Cloud mode uses this ID. In direct connection mode, a non-`null` ID enables the OAuth credentials provider, which uses the secret and audience. The Connector does not validate credential pairing itself; prepare credentials according to the gateway's authentication requirements.

#### `zeebe.client.cloud.clientSecret`

Specifies the OAuth client secret.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Cloud mode uses this value; direct connections use it only when a client ID is set. It is registered as `STRING`, not `PASSWORD`, so do not assume that configuration displays automatically redact it. Supply it through the platform's secret management mechanism, and do not expose it in logs or shared configurations. The property name is `clientSecret`; do not replace it with another SDK property name.

#### `zeebe.client.cloud.token.audience`

Specifies the token audience for direct-connection OAuth credentials.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Used only when the cloud cluster ID is unset and the client ID is set. Enter the audience required by the authentication system. Cloud mode ignores this setting, and `null` does not represent a fixed audience string.

#### `zeebe.client.cloud.authorization.server.url`

Specifies an authorization server URL, but this version does not use the setting.

* **Type**: `string`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: In Connector 0.51.0, this setting is not applied when building the client, so changing it cannot configure the token retrieval endpoint. Do not treat it as a usable setting for a custom OAuth authorization URL.

### Job Selection and Routing

#### `job.types`

Selects the Zeebe Job types to activate.

* **Type**: `list`
* **Default**: `kafka`
* **Importance**: High
* **Valid Values / Notes**: A comma-separated list of types that must not be empty after parsing. Matches the Job types of service tasks; it does not support regular expressions and is not a process event type filter. The list is not automatically deduplicated; use distinct types.

#### `job.header.topics`

Specifies the Zeebe custom header key containing the destination Kafka Topic name.

* **Type**: `string`
* **Default**: `kafka-topic`
* **Importance**: High
* **Valid Values / Notes**: The configuration value is the header key, not a Topic name. Each Job must provide a nonempty, valid, single Topic name under this key. The entire header value is used directly as the destination Topic; it is not split on commas and does not perform multi-Topic fan-out. A nonempty check alone does not establish that a Kafka name is valid.

#### `job.variables`

Selects the process variables to fetch when activating jobs.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: Low
* **Valid Values / Notes**: Use comma-separated variable names; an empty list fetches all variables. This setting only narrows the variables fetched. It does not filter Jobs or remove the outer job metadata and custom headers.

### Job Activation and Timeouts

#### `zeebe.client.worker.name`

Specifies the Zeebe worker name used when claiming jobs.

* **Type**: `string`
* **Default**: `kafka-connector`
* **Importance**: Low
* **Valid Values / Notes**: Passed unchanged to the job activation command. It is not the Kafka Connect Worker name and does not establish an exclusive lock or job sharding.

#### `zeebe.client.worker.maxJobsActive`

Sets the target capacity for activated and in-flight jobs per Task, per Job type.

* **Type**: `int`
* **Default**: `100`
* **Importance**: Medium
* **Valid Values / Notes**: ConfigDef declares no range validation; use a positive integer. Zero or negative values provide no activation capacity. This is not a strict in-flight limit for the entire Connector or a global batch size for one poll. When adjusting it, observe actual operation alongside write latency, backlog, and Worker resources.

#### `zeebe.client.job.timeout`

Sets the lock duration after Job activation, in milliseconds.

* **Type**: `long`
* **Default**: `5000`
* **Importance**: Medium
* **Valid Values / Notes**: ConfigDef declares no range validation; use a positive duration that covers actual Kafka write and Zeebe completion latency. Unfinished jobs can be activated again after the timeout, and the Connector does not automatically renew the lock. This is neither the activation RPC request timeout nor the Connect offset commit timeout.

## Best Practices

### Define Message Content During Initial Integration

**Business Scenario**: The process already has a message-sending step, and downstream consumers need to be connected. Before connecting them, define the variables and message encoding they need instead of exposing all process variables by default.

**Configuration Example**: Add the following setting to the Quick Start configuration. `orderId` and `status` are variable names in the example process; replace them with variables that consumers actually need and that the process already provides.

```properties theme={null}
job.variables=orderId,status
```

**Key Notes**: First confirm the process type and Topic header, then use one process instance to check the message's variables, Job key, and outer job content before connecting consumers. Selecting variables does not change the outer structure of the Job JSON. Consumers should retain the Job key to identify repeated processing. If the business needs all variables, leave the list empty; there is no need to narrow the scope just to use this example. The Connector completes service tasks and should not inadvertently compete with other workers responsible for business processing of the same task.

### Allow Time for Failure Recovery During Normal Operation

**Business Scenario**: The Connector is running continuously, but slower Kafka writes, brief network interruptions, or Task restarts may cause jobs to be claimed again. Normal processing needs enough time, and consumers need to recognize duplicate jobs.

**Configuration Example**: Add the following setting to the Quick Start configuration. `30000` milliseconds is an example adjustment; determine the value from actual write and completion latency.

```properties theme={null}
zeebe.client.job.timeout=30000
```

**Key Notes**: The lock duration should cover normal processing latency with a margin. A duration that is too short increases opportunities for redelivery; one that is too long delays reclaiming unfinished jobs. Consumers should use the Job key as an idempotency identifier when handling duplicate records for the same job. Recovery after a restart depends on Zeebe job state and timeouts, not on locating historical records through Connect offsets. Avoid SMTs or error-tolerance policies that unintentionally discard records: filtered records or records discarded under error tolerance may still trigger job completion. Completion commands are sent asynchronously without waiting for results, and Kafka writes and job completion are not atomic. Neither increasing the lock duration nor consumer deduplication provides end-to-end loss-free or exactly-once guarantees.

### Increase Parallel Processing Across Existing Job Types

**Business Scenario**: The integration expands from one job type to multiple existing message-sending job types, and sequential activation of different types by a single Task has begun to affect processing latency. Assign the different types to different Tasks to avoid adding idle Tasks that cannot handle those jobs.

**Configuration Example**: Add the following settings to the Quick Start configuration. The example process already uses the Job types `order-events` and `shipment-events`; jobs of both types must have valid Topic headers.

```properties theme={null}
job.types=order-events,shipment-events
tasks.max=2
```

**Key Notes**: This overrides the default `kafka` type; only the listed types are claimed. Before expanding, confirm that all target types are in the list and observe Task assignment, throughput, and latency. Reconfiguration may regroup types. Do not increase the Task count beyond the number of distinct types solely to meet throughput needs; one busy type is not automatically split across multiple Tasks. Do not duplicate list entries or redesign business job types just to increase parallelism. More Tasks may increase the total number of activated jobs, so also monitor resources and job backlog. Global ordering across types, Tasks, or Kafka partitions is not guaranteed.

## Monitoring

### What to Monitor

Monitor Kafka Connect Connector / Task status, throughput and end-to-end latency, offset commit duration and failures, errors and retries, and Worker JVM heap memory, GC, and thread signals. A Task in RUNNING state does not necessarily mean it has successfully claimed or published jobs; assess actual messages and logs as well. Monitor DLQ activity only when the deployment enables the corresponding error handling, and do not treat DLQ activity as proof of Zeebe job completion.

### Import the Grafana Dashboard

Download the shared [Kafka Connect Grafana Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), confirm that Kafka Connect metrics are collected into a compatible Prometheus data source and that collection labels match the cluster, Connector, and Task labels used by the dashboard, then import the JSON into Grafana and select the corresponding data source.

## Limitations

* The data input consists of activatable Jobs of the specified types. Exporting all process events, replaying historical records, or filtering jobs by process ID, element, tenant, or arbitrary conditions is not supported.
* Each Job routes to only one Topic; comma-separated strings in custom headers are not expanded into multiple destination Topics.
* The Connector does not automatically split a single distinct Job type across multiple Tasks. Increasing `tasks.max` does not directly improve parallelism for a single type.
* Kafka writes and Zeebe job completion do not form a cross-system atomic transaction. Completion commands do not wait for asynchronous acknowledgment; the Connector provides neither end-to-end exactly-once nor unconditional at-least-once delivery guarantees.
* Connect offsets contain the Job key but are not used to locate or filter jobs at startup. Resetting offsets does not republish completed Jobs.
* The Connector does not automatically renew activation locks or guarantee that all asynchronous completion commands have been acknowledged when it stops.
* `zeebe.client.cloud.authorization.server.url` has no effect in Connector 0.51.0 and cannot configure a custom token retrieval endpoint.

## FAQ

### Why Are There No Messages in the Topic When the Task Shows RUNNING?

Check whether the process has reached an activatable service task, whether its Job type matches `job.types`, and whether other workers are competing for the same type. Then check the gateway address, TLS and authentication modes, and activation error logs; persistent authorization errors may also appear as empty polls. Confirm that the header points to a single valid Topic, that the Topic exists and permits writes, and check for Converter and Kafka write errors. Running status alone does not establish successful connectivity or publication.

### Why Does a Missing Topic Header Cause a Job Failure or an Incident?

When a job lacks the specified header or its value is empty, the Connector produces no output record. Instead, it asynchronously sends a job failure command that decreases the remaining retry count by one. If Zeebe accepts the failure command, jobs with retries remaining can be claimed again; exhausting the retries produces an incident. Check that the key specified by `job.header.topics` matches the process definition and correct the Topic header in the process. Handle existing failed jobs through Zeebe incident and retry management. A nonempty but invalid Topic name may pass the local check and fail during Kafka writes. Check Kafka errors rather than expecting the missing-header handling path to address it.

### Why Do Two Topic Names Not Produce Two Messages?

The header value is not split on commas; the entire value is treated as one Topic name. Change the header to a single valid Topic. If the business needs multiple destinations, design separate message-sending steps in the workflow or implement distribution downstream in Kafka rather than relying on this Connector's header for fan-out.

### Why Does the Message Contain Job Metadata Instead of Just Variables?

The message value is the activated Job's JSON, including the outer job content. `job.variables` only controls which variables are fetched; it does not strip the outer structure. Parse the message according to its actual encoding and extract the required variables. When using JsonConverter, distinguish its wrapping of the string from the Job JSON itself. To read JSON text directly, use StringConverter as in the Quick Start.

### Why Does the Process Not Continue Even Though Kafka Has a Message?

Job completion commands are sent asynchronously, and the Connector neither waits for nor checks their asynchronous results. Successful Kafka writes do not prove that Zeebe has acknowledged completion. Check the job's actual state in Zeebe, gateway connectivity and authentication, and whether the job has timed out and been reactivated. Restore connectivity and handle jobs that remain unfinished, reconciling published messages with process state according to business requirements. Do not reset Connect offsets as a process repair measure.

### Why Does the Same Job Key Reappear After a Restart, While Resetting Offsets Does Not Replay Jobs?

If a failure or lock timeout occurs after a Kafka write but before Zeebe completion acknowledgment, an unfinished job may be activated again with the same Job key. Consumers should implement idempotent handling using that key and adjust the Job lock duration according to actual latency. Connect offsets are not read as replayable cursors; completed Jobs are not reactivated by offset resets. To resend business messages, use business compensation or new process tasks.

### Why Does Changing the Authorization Server URL Not Change the Authentication Endpoint?

Although Connector 0.51.0 registers `zeebe.client.cloud.authorization.server.url`, it does not apply it to the client. Confirm whether cloud mode or direct connection mode is in use and check the applicable client ID, secret, and audience. Repeatedly changing this setting cannot switch the token endpoint. If the deployment requires a custom authorization endpoint, choose an integration that explicitly supports that authentication method, or confirm support in the target Connector before upgrading.