> Part of the walkerOS documentation. Project overview and full index: <https://www.walkeros.io/llms.txt>

# Apache Kafka

<!-- -->

[Server](#)[ ](https://github.com/elbwalker/walkerOS/tree/main/packages/server/destinations/kafka)

<!-- -->

[Source code](https://github.com/elbwalker/walkerOS/tree/main/packages/server/destinations/kafka)[ ](https://www.npmjs.com/package/@walkeros/server-destination-kafka)

<!-- -->

[Package](https://www.npmjs.com/package/@walkeros/server-destination-kafka)

<!-- -->

Beta

Server-side event streaming to [Apache Kafka](https://kafka.apache.org/) via the [`kafkajs`](https://kafka.js.org/) client. Each event is serialized as JSON, keyed for partition-friendly ordering, and produced to a configurable topic. Supports SASL/SSL authentication (Confluent Cloud, AWS MSK, SCRAM), configurable compression (gzip, snappy, lz4, zstd), per-rule topic and key overrides, and graceful shutdown via `destroy()`.

<!-- -->

Where this fits

Kafka is a **server destination** in the walkerOS flow:

Receives events server-side from the collector, serializes them as JSON with a partition-friendly message key, and produces them to a Kafka topic for downstream stream processing (Flink, Spark, ksqlDB, Kafka Connect, consumers).

## Installation[​](#installation "Direct link to Installation")

```
npm install @walkeros/server-destination-kafka
```

* Integrated
* Bundled

```
import { startFlow } from '@walkeros/collector';
import { destinationKafka } from '@walkeros/server-destination-kafka';

await startFlow({
  destinations: {
    kafka: {
      code: destinationKafka,
      config: {
        settings: {
          kafka: {
            brokers: ['kafka1:9092', 'kafka2:9092'],
            topic: 'walkeros-events',
          },
        },
      },
    },
  },
});
```

Add to your `flow.json` destinations:

```
"destinations": {
  "kafka": {
    "package": "@walkeros/server-destination-kafka",
    "config": {
      "settings": {
        "kafka": {
          "brokers": ["kafka1:9092", "kafka2:9092"],
          "topic": "walkeros-events"
        }
      }
    }
  }
}
```

[See bundled mode setup](https://www.walkeros.io/docs/getting-started/modes/bundled.md) | [CLI reference](https://www.walkeros.io/docs/apps/cli.md)

## Configuration[​](#configuration "Direct link to Configuration")

This <!-- -->destination<!-- --> uses the standard <!-- -->destination<!-- --> config wrapper (consent, data, env, id, ...). For the shared fields see [destination<!-- --> configuration](https://www.walkeros.io/docs/destinations.md#configuration). Package-specific fields live under `config.settings` and are listed below.

## Settings[​](#settings "Direct link to Settings")

| Property                 | Type                                | Description                                                                                         | More |
| ------------------------ | ----------------------------------- | --------------------------------------------------------------------------------------------------- | ---- |
| `kafka*`                 | `kafka`                             | Kafka connection and producer settings.                                                             |      |
| `brokers*`               | `Array<string>`                     | Kafka broker addresses (host:port). At least one required.                                          |      |
| `clientId`               | `string`                            | Kafka client ID. Default: walkeros.                                                                 |      |
| `ssl`                    | `boolean \| object`                 | TLS configuration. Set true for default TLS, or provide a tls.ConnectionOptions object for mTLS.    |      |
| `sasl`                   | `sasl`                              | SASL authentication config. Required for Confluent Cloud, AWS MSK with IAM, etc.                    |      |
| `mechanism*`             | `'plain' \| 'scram-sha-256' \| 's`  | SASL authentication mechanism.                                                                      |      |
| `username`               | `string`                            | Username for plain/scram mechanisms.                                                                |      |
| `password`               | `string`                            | Password for plain/scram mechanisms.                                                                |      |
| `accessKeyId`            | `string`                            | AWS access key ID for IAM auth (mechanism: aws).                                                    |      |
| `secretAccessKey`        | `string`                            | AWS secret access key for IAM auth (mechanism: aws).                                                |      |
| `sessionToken`           | `string`                            | AWS session token for temporary credentials (mechanism: aws).                                       |      |
| `authorizationIdentity`  | `string`                            | AWS authorization identity (mechanism: aws).                                                        |      |
| `connectionTimeout`      | `integer`                           | Connection timeout in ms. Default: 1000.                                                            |      |
| `requestTimeout`         | `integer`                           | Request timeout in ms. Default: 30000.                                                              |      |
| `topic*`                 | `string`                            | Target Kafka topic name.                                                                            |      |
| `acks`                   | `integer`                           | Acknowledgement level. -1 = all replicas, 0 = fire-and-forget, 1 = leader only. Default: -1.        |      |
| `timeout`                | `integer`                           | Broker response timeout in ms. Default: 30000.                                                      |      |
| `compression`            | `'none' \| 'gzip' \| 'snappy' \| '` | Message compression codec. Default: gzip. Snappy/LZ4/ZSTD require additional npm packages.          |      |
| `idempotent`             | `boolean`                           | Enable idempotent producer for exactly-once delivery. Default: false.                               |      |
| `allowAutoTopicCreation` | `boolean`                           | Allow auto-creation of topics on the broker. Default: false.                                        |      |
| `key`                    | `string`                            | Mapping value path for message key derivation (e.g. user.id, data.userId). Default: entity\_action. |      |
| `headers`                | `Record<string, string>`            | Static headers added to every message.                                                              |      |
| `retry`                  | `retry`                             | Retry configuration for transient failures.                                                         |      |
| `maxRetryTime`           | `integer`                           | Max total retry wait in ms. Default: 30000.                                                         |      |
| `initialRetryTime`       | `integer`                           | First retry delay in ms. Default: 300.                                                              |      |
| `retries`                | `integer`                           | Max retry count. Default: 5.                                                                        |      |

\* Required fields

## Mapping[​](#mapping "Direct link to Mapping")

Per-event rules under `config.mapping`. For the standard rule fields (consent, condition, data, batch, name, policy) see [mapping](https://www.walkeros.io/docs/mapping.md).

| Property | Type     | Description                                                                                               | More |
| -------- | -------- | --------------------------------------------------------------------------------------------------------- | ---- |
| `key`    | `string` | Override message key mapping path for this rule (e.g. data.id). Takes precedence over settings.kafka.key. |      |
| `topic`  | `string` | Override Kafka topic for this rule. Takes precedence over settings.kafka.topic.                           |      |

## Examples

### Default event

An event is produced to the configured Kafka topic with the full JSON body and entity\_action as the message key.

Event

```
{
  "name": "page view",
  "data": {
    "domain": "www.example.com",
    "title": "walkerOS documentation",
    "referrer": "https://www.walkeros.io/",
    "search": "?foo=bar",
    "hash": "#hash",
    "id": "/docs/"
  },
  "context": {
    "dev": [
      "test",
      1
    ]
  },
  "globals": {
    "pagegroup": "docs"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "child",
      "data": {
        "is": "subordinated"
      }
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "09b9da5a9f4c5500",
  "trigger": "load",
  "entity": "page",
  "action": "view",
  "timestamp": 1700000100,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}
```

Out

```
producer.send({
  "topic": "walkeros-events",
  "messages": [
    {
      "key": "page_view",
      "value": "json:event",
      "headers": {
        "content-type": "application/json"
      },
      "timestamp": "1700000100"
    }
  ],
  "acks": -1,
  "compression": 1
})
```

### Key from user id

A settings.kafka.key path resolves the message key from the event, here using user.id for per-user partitioning.

Event

```
{
  "name": "user signup",
  "data": {
    "plan": "pro"
  },
  "context": {
    "dev": [
      "test",
      1
    ]
  },
  "globals": {
    "lang": "elb"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "usr-789"
  },
  "nested": [
    {
      "entity": "child",
      "data": {
        "is": "subordinated"
      }
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "19d6e6b339fcfe62",
  "trigger": "test",
  "entity": "user",
  "action": "signup",
  "timestamp": 1700000103,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}
```

Out

```
producer.send({
  "topic": "walkeros-events",
  "messages": [
    {
      "key": "usr-789",
      "value": "json:event",
      "headers": {
        "content-type": "application/json"
      },
      "timestamp": "1700000103"
    }
  ],
  "acks": -1,
  "compression": 1
})
```

### Mapped payload

A data mapping transforms the event payload before producing it as the Kafka message value.

Event

```
{
  "name": "order complete",
  "data": {
    "id": "ORD-400",
    "total": 99.99,
    "currency": "EUR"
  },
  "context": {
    "shopping": [
      "complete",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "product",
      "data": {
        "id": "ers",
        "name": "Everyday Ruck Snack",
        "color": "black",
        "size": "l",
        "price": 420
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "product",
      "data": {
        "id": "cc",
        "name": "Cool Cap",
        "size": "one size",
        "price": 42
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "gift",
      "data": {
        "name": "Surprise"
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "a0e4edd473406dff",
  "trigger": "load",
  "entity": "order",
  "action": "complete",
  "timestamp": 1700000102,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}
```

Mapping

```
{
  "name": "purchase",
  "data": {
    "map": {
      "order_id": "data.id",
      "revenue": "data.total",
      "currency": "data.currency"
    }
  }
}
```

Out

```
producer.send({
  "topic": "walkeros-events",
  "messages": [
    {
      "key": "purchase",
      "value": "json:data",
      "headers": {
        "content-type": "application/json"
      },
      "timestamp": "1700000102"
    }
  ],
  "acks": -1,
  "compression": 1
})
```

### Renamed event

A mapping renames the event which also changes the default Kafka message key used for partitioning.

Event

```
{
  "name": "order complete",
  "data": {
    "id": "0rd3r1d",
    "currency": "EUR",
    "shipping": 5.22,
    "taxes": 73.76,
    "total": 555
  },
  "context": {
    "shopping": [
      "complete",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "product",
      "data": {
        "id": "ers",
        "name": "Everyday Ruck Snack",
        "color": "black",
        "size": "l",
        "price": 420
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "product",
      "data": {
        "id": "cc",
        "name": "Cool Cap",
        "size": "one size",
        "price": 42
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "gift",
      "data": {
        "name": "Surprise"
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "c89e881431a6b3b2",
  "trigger": "load",
  "entity": "order",
  "action": "complete",
  "timestamp": 1700000101,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}
```

Mapping

```
{
  "name": "purchase"
}
```

Out

```
producer.send({
  "topic": "walkeros-events",
  "messages": [
    {
      "key": "purchase",
      "value": "json:event",
      "headers": {
        "content-type": "application/json"
      },
      "timestamp": "1700000101"
    }
  ],
  "acks": -1,
  "compression": 1
})
```

### Topic override

A mapping rule overrides the destination topic so specific events are routed to a dedicated stream.

Event

```
{
  "name": "order complete",
  "data": {
    "id": "ORD-500",
    "total": 42
  },
  "context": {
    "shopping": [
      "complete",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "product",
      "data": {
        "id": "ers",
        "name": "Everyday Ruck Snack",
        "color": "black",
        "size": "l",
        "price": 420
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "product",
      "data": {
        "id": "cc",
        "name": "Cool Cap",
        "size": "one size",
        "price": 42
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "gift",
      "data": {
        "name": "Surprise"
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "4c2fe718946eae38",
  "trigger": "load",
  "entity": "order",
  "action": "complete",
  "timestamp": 1700000104,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "collector",
    "schema": "4"
  }
}
```

Mapping

```
{
  "settings": {
    "topic": "orders-stream"
  }
}
```

Out

```
producer.send({
  "topic": "orders-stream",
  "messages": [
    {
      "key": "order_complete",
      "value": "json:event",
      "headers": {
        "content-type": "application/json"
      },
      "timestamp": "1700000104"
    }
  ],
  "acks": -1,
  "compression": 1
})
```

The destination creates a single long-lived `kafkajs` producer during `init()` and calls `producer.connect()` before accepting events. On flow hot-swap or server shutdown, `destroy()` calls `producer.disconnect()` to flush in-flight messages and close TCP connections.

## Message format[​](#message-format "Direct link to Message format")

Events are serialized as JSON and produced with the following structure:

* **topic**, from `settings.kafka.topic` (or `mapping.settings.topic` override)
* **key**, resolved from `settings.kafka.key` (or `mapping.settings.key` override) mapping path; defaults to the event name with spaces replaced by `_` (e.g. `page_view`, `order_complete`) for partition-based ordering
* **value**, mapped payload (when `data.map` is configured) or the full walkerOS event, `JSON.stringify()`-ed
* **headers**, `content-type: application/json` plus any static `settings.kafka.headers`
* **timestamp**, event timestamp as string (ms since epoch)

Use `mapping.settings.topic` to route specific events to dedicated topics (e.g. orders to `orders-stream`, identities to `identity-stream`). Use `mapping.settings.key` to set a key path per rule (e.g. `data.order_id` for order events).

## Setup[​](#setup "Direct link to Setup")

No safe defaults

Kafka topic creation requires explicit `numPartitions` and `replicationFactor`. There is no universally correct default for either: `replicationFactor` must be less than or equal to broker count, and `numPartitions` is a function of expected throughput and consumer parallelism. The boolean form `setup: true` is rejected with an error listing the required fields. Only the object form is valid.

Provision a topic once per environment with the CLI:

```
walkeros setup destination.kafka
```

Output: `setup: ok destination.kafka` plus a JSON line reporting `{ topicCreated, schemaRegistered }`. The command is idempotent, safe to re-run. Drift on `numPartitions`, `replicationFactor`, or `configEntries` is logged as `WARN setup.drift {...}` and never auto-mutates the broker.

`config.setup`:

* `false` (default): no provisioning. Operator must run setup explicitly.
* `true`: rejected at runtime with an actionable error. There are no safe defaults for partition count or replication factor.
* `{ numPartitions, replicationFactor, ... }`: object form is the only valid form. See the `Setup` interface in the package for full options.

### Required fields[​](#required-fields "Direct link to Required fields")

| Field               | Type   | Notes                                              |
| ------------------- | ------ | -------------------------------------------------- |
| `numPartitions`     | number | Required at runtime. No safe default.              |
| `replicationFactor` | number | Required at runtime. Must be `<= broker count`.    |
| `topic`             | string | Falls back to `settings.kafka.topic` when omitted. |

### Optional fields[​](#optional-fields "Direct link to Optional fields")

| Field            | Type    | Notes                                                       |
| ---------------- | ------- | ----------------------------------------------------------- |
| `configEntries`  | object  | Topic-level config, e.g. `{ "retention.ms": "604800000" }`. |
| `schemaRegistry` | object  | Confluent Schema Registry binding (see below).              |
| `validateOnly`   | boolean | kafkajs broker-side dry-run. No topic is created.           |

### Example[​](#example "Direct link to Example")

```
{
  "destinations": {
    "kafka": {
      "package": "@walkeros/server-destination-kafka",
      "config": {
        "settings": {
          "kafka": { "brokers": ["broker:9092"], "topic": "walkeros-events" }
        },
        "setup": {
          "numPartitions": 6,
          "replicationFactor": 3,
          "configEntries": { "retention.ms": "604800000" }
        }
      }
    }
  }
}
```

### Schema Registry (optional)[​](#schema-registry-optional "Direct link to Schema Registry (optional)")

```
{
  "setup": {
    "numPartitions": 6,
    "replicationFactor": 3,
    "schemaRegistry": {
      "url": "https://schema-registry.example.com",
      "subject": "walkeros-events-value",
      "schemaType": "JSON",
      "schema": "{ \"type\": \"object\" }",
      "compatibility": "BACKWARD"
    }
  }
}
```

The schema is registered via the Confluent Schema Registry REST API. The optional `compatibility` level is set on the subject after registration.

### Runtime error when the topic is missing[​](#runtime-error-when-the-topic-is-missing "Direct link to Runtime error when the topic is missing")

When `setup` was not run and the topic does not exist on the cluster, `push()` catches the kafkajs `UNKNOWN_TOPIC_OR_PARTITION` error and logs an actionable message pointing the operator at `walkeros setup destination.<id>`. Run setup with explicit `numPartitions` and `replicationFactor` to provision the topic.

## Authentication[​](#authentication "Direct link to Authentication")

### Confluent Cloud (SASL/PLAIN)[​](#confluent-cloud-saslplain "Direct link to Confluent Cloud (SASL/PLAIN)")

```
{
  "kafka": {
    "brokers": ["pkc-xxxxx.us-east-1.aws.confluent.cloud:9092"],
    "topic": "walkeros-events",
    "ssl": true,
    "sasl": {
      "mechanism": "plain",
      "username": "$env.CONFLUENT_API_KEY",
      "password": "$env.CONFLUENT_API_SECRET"
    }
  }
}
```

### AWS MSK (IAM)[​](#aws-msk-iam "Direct link to AWS MSK (IAM)")

```
{
  "kafka": {
    "brokers": ["broker.msk.us-east-1.amazonaws.com:9098"],
    "topic": "walkeros-events",
    "ssl": true,
    "sasl": {
      "mechanism": "aws",
      "accessKeyId": "$env.AWS_ACCESS_KEY",
      "secretAccessKey": "$env.AWS_SECRET_KEY"
    }
  }
}
```