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

# Kafka Destination Connector: Setup and Configuration

> Publish change events from your source database to your own Apache Kafka or Amazon MSK cluster. Artie writes one topic per table as Debezium JSON, with SASL, AWS MSK IAM, TLS, and mutual TLS support.

> [!NOTE]
>
> Kafka destinations are turned on per account. Contact Artie to enable them for yours.

## What you'll need

* An Apache Kafka or Amazon MSK cluster.
* Network access from Artie to every broker the cluster advertises, not only the bootstrap servers. Artie connects directly, so allow [Artie's IP addresses](/docs/connection-options.md#fixed-ip-addresses). SSH tunnels and PrivateLink aren't supported for Kafka.
* A Kafka principal that can write to Artie's topics. You create it in the first step.
* If your cluster doesn't create topics automatically, either create each table's topic before the pipeline writes to it, or turn on **Create missing topics**. That setting needs Kafka 2.4 or later and permission to create topics.

## How Artie writes to Kafka

Artie's reader publishes each change straight to your cluster. There's no separate writer between your source and your topics, so settings that Artie's writer applies to warehouse destinations don't apply here. See [Limitations](#limitations).

Each table gets its own topic, named `<topic prefix>.<table identifier>`. The table identifier depends on the source, for example:

| Source | Topic name |
| - | - |
| PostgreSQL | `<prefix>.<schema>.<table>`. Partitions publish to their parent table's topic. |
| MySQL | `<prefix>.<database>.<table>` |
| Microsoft SQL Server | `<prefix>.<database>.<schema>.<table>` |
| MongoDB | `<prefix>.<database>.<collection>` |

Tables that unify across schemas publish to `<prefix>._.<table>`. SQL Server tables that unify across databases publish to `<prefix>.<schema>.<table>`.

## Setup

**1. Give Artie access to the cluster**

Artie writes to topics that start with your topic prefix. The examples below use the prefix `artie`. Replace it with your own.

**SASL/SCRAM or PLAIN**

Create a user for Artie, then allow it to describe and write to topics with your prefix. Kafka's producer writes idempotently, so on Kafka versions before 3.0 the user also needs `IdempotentWrite` on the cluster.

```bash theme={null}
kafka-configs.sh --bootstrap-server b-1.kafka.example.com:9096 --command-config admin.properties \
  --alter --add-config 'SCRAM-SHA-512=[password=s3cure-Passw0rd]' \
  --entity-type users --entity-name artie

kafka-acls.sh --bootstrap-server b-1.kafka.example.com:9096 --command-config admin.properties \
  --add --allow-principal User:artie \
  --operation Describe --operation Write \
  --topic artie --resource-pattern-type prefixed
```

To let Artie create missing topics, also allow `Create`:

```bash theme={null}
kafka-acls.sh --bootstrap-server b-1.kafka.example.com:9096 --command-config admin.properties \
  --add --allow-principal User:artie \
  --operation Create \
  --topic artie --resource-pattern-type prefixed
```

**AWS MSK IAM**

Create an IAM role in the AWS account that owns the cluster. Artie assumes this role to sign in. Attach a policy that lets it connect to the cluster and write to topics with your prefix. Replace the Region, account ID, and cluster name with your own.

```json theme={null}
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:Connect",
        "kafka-cluster:WriteDataIdempotently"
      ],
      "Resource": "arn:aws:kafka:us-east-1:123456789012:cluster/orders-cluster/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:DescribeTopic",
        "kafka-cluster:WriteData"
      ],
      "Resource": "arn:aws:kafka:us-east-1:123456789012:topic/orders-cluster/*/artie.*"
    }
  ]
}
```

To let Artie create missing topics, add `kafka-cluster:CreateTopic` to the topic statement.

The role's trust policy must allow the IAM role that Artie's data plane runs as to assume it. Contact Artie for that role's ARN. If you also set an **External ID**, add a `sts:ExternalId` condition with the same value. On a data plane you run yourself, trust the IAM identity your data plane runs as.

**2. Add the Kafka connector in Artie**

Create a destination connector, choose **Kafka**, and fill in:

| Setting | What to enter |
| - | - |
| **Bootstrap servers** | A comma-separated `host:port` list, for example `b-1.kafka.example.com:9096,b-2.kafka.example.com:9096`. |
| **Topic prefix** | The start of every topic name. Letters, digits, `.`, `_`, and `-` only. |
| **Authentication** | **SCRAM-SHA-512**, **SCRAM-SHA-256**, **PLAIN**, **AWS MSK IAM**, or **None**. |
| **Username** and **Password** | For SCRAM and PLAIN. |
| **AWS role ARN** | For AWS MSK IAM. The role from the previous step. |
| **External ID (optional)** | For AWS MSK IAM, if the role's trust policy requires one. |
| **Disable TLS** | Only for brokers with a plaintext listener. Not available with AWS MSK IAM, which always uses TLS. |
| **CA certificate (optional)** | A PEM bundle Artie uses instead of the system roots to verify the brokers. |
| **Client certificate (optional)** and **Client key (optional)** | A PEM pair for mutual TLS. A client certificate needs its key. |
| **Create missing topics** | Turn this on if your cluster doesn't create topics automatically. See [Topic creation](#topic-creation). |

When you test the connection, Artie signs in to a bootstrap server, then checks that it can reach every broker the cluster advertises. Producers write to each partition's leader, so an unreachable broker would fail writes later. If one isn't reachable, the test fails with `broker <id> at <host>:<port> is unreachable`. Check your firewall rules and the brokers' advertised listeners.

**3. Create the pipeline**

Choose your source and the Kafka connector, then pick the tables to publish. Artie backfills each table into its topic, then publishes changes as they happen.

## Advanced

> **Message format**
>
> Each message is one change to one row, as [Debezium](https://debezium.io/documentation/reference/stable/connectors/postgresql.html#postgresql-events)-style JSON.
>
> The key holds the row's primary key, so every change to a row goes to the same partition, in order:
>
> ```json theme={null}
> {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"default":null,"field":"id","name":"","parameters":null}],"optional":false,"field":""},"payload":{"id":1042}}
> ```
>
> The value has a `schema` that describes each column's type, and a `payload` with the change:
>
> ```json theme={null}
> {
>   "payload": {
>     "before": null,
>     "after": {"id": 1042, "status": "pending"},
>     "source": {"connector": "reader", "ts_ms": 1792537200000, "db": "shop", "schema": "public", "table": "orders"},
>     "op": "c"
>   }
> }
> ```
>
> * `op` is `c` for inserts, `u` for updates, `d` for deletes, and `r` for rows from a backfill.
> * For a delete, `after` is `null` and `before` holds the deleted row.
> * `source.ts_ms` is when the change happened in the source, in milliseconds since the Unix epoch.
> * Values use Debezium's encodings, which the `schema` describes for each column.
>
> Artie compresses batches with LZ4, which Kafka clients decompress for you.

> **Topic creation**
>
> Kafka creates topics automatically when `auto.create.topics.enable` is on, which is the default for self-managed Kafka.
>
> If it's off, a write to a topic that doesn't exist fails. Without **Create missing topics**, Artie keeps retrying and the pipeline reports an error until the topic exists. Create topics before you add tables to the pipeline, including tables Artie adds automatically later.
>
> With **Create missing topics** on, when a write fails because its topic doesn't exist, Artie creates the topic with your cluster's default partitions, replication factor, and retention, then retries. Topics that already exist are never touched. The first write to a new topic takes a few extra seconds. To use other settings, create the topic yourself before the pipeline writes to it.
>
> Creating a topic needs permission to create topics and Kafka 2.4 or later. If the broker refuses, for example because the permission is missing, the pipeline reports the broker's error.

> **Delivery guarantees and retries**
>
> Delivery is at least once. Make your consumers idempotent, for example by upserting on the message key.
>
> * If a publish fails, Artie retries the messages that failed up to 10 times, waiting up to 5 seconds between attempts. Messages that were already accepted aren't sent again.
> * If every attempt fails, the pipeline reports an error. Artie only advances its position in your source after Kafka accepts the messages, so a failed publish doesn't lose changes.
> * After an error or a restart, Artie publishes again from its last committed position, so consumers can see some messages twice.

> **Message size**
>
> Artie sends messages up to 1 MiB. A message larger than that isn't published: Artie skips it, sends a [row skipped](/docs/api/webhooks/row-skipped.md) webhook event with the table and primary key, and keeps going. Make sure your topics accept messages of at least 1 MiB (the topic's `max.message.bytes`, or the broker's `message.max.bytes`).
>
> > [!WARNING]
> >
> > Skipped messages don't reach your topics. If a table has rows larger than 1 MiB, contact Artie before adding it to a Kafka pipeline.
>
> With the pipeline's [message compression](/docs/pipelines/overview.md#message-compression) set to gzip, Artie gzips values of 500 KiB or more before checking their size, so fewer rows are skipped. Those messages carry the header `artie-compression: gzip`. Consumers must check for that header and decompress the value before parsing it.

> **Backfills**
>
> Kafka topics can't be emptied before a backfill. When you backfill or restart a backfill, Artie publishes every row again on top of the messages already in the topic, with `op` set to `r`. Consumers see each backfilled row again, so handle `r` like an upsert.

> **Monitoring and billing**
>
> Artie counts the rows it publishes. Pipeline monitors and charts for a Kafka destination use those counts and the time from a change in your source to its publish.
>
> Billing counts published inserts, updates, and deletes. Rows from backfills (`op` `r`) aren't billed.

## Limitations

* **No warehouse writer settings.** These don't apply to Kafka destinations and are hidden in the Dashboard: history mode, column hashing and encryption, soft deletes, dropping deleted columns, casting columns to strings, the Additional data, Data handling, and Optimizations settings, and flush rules. **Auto-replicate new tables** and **Disable alert emails** still apply.
* **No table-level monitors.** Saving one fails with `Table monitors aren't supported for Kafka destinations. Monitor the whole pipeline instead.` Monitor the whole pipeline.
* **Topics can't be emptied, dropped, or swapped before a backfill.** See [Backfills](#backfills).
* **Messages over 1 MiB are skipped.** See [Message size](#message-size).
* **No schema change messages.** Artie publishes row changes only. A column's type and presence show up in each message's `schema`.
* **Direct connections only.** SSH tunnels and PrivateLink aren't supported. Saving a Kafka connector with PrivateLink fails with `PrivateLink isn't supported for Kafka`.

## Frequently asked questions

### Why do my consumers see the same message twice?

A publish was retried, the pipeline restarted, or a table was backfilled again. Delivery is at least once, so make consumers idempotent. See [Delivery guarantees and retries](#delivery-guarantees-and-retries).

### Why is the pipeline failing with an unknown topic error?

The topic doesn't exist and your cluster doesn't create topics automatically. Create the topic, or turn on **Create missing topics**. See [Topic creation](#topic-creation).

### How do I tell a delete from an update?

Read `op` in the message value. It's `d` for deletes, where `before` holds the deleted row, and `u` for updates.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.