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

# Configure event sources: Kafka inbound consume

> Configure the inbound Kafka event source — brokers, topic, consumer group, type mapping, and encrypted credentials — plus a dry-run consume-batch test and failure-model notes.

# Configure event sources: Kafka inbound consume

An event **source** is the inbound mirror of an [event sink](/guides/event-sinks-streaming): the sink streams the platform's own event taxonomy **out** to your Kafka topic or HTTP collector, and the source consumes your **own** Kafka topic **into** Orbit. If you run Kafka as your event system of record, an inbound source lets Orbit subscribe to your domain events — orders, payments, fraud signals, CRM updates — and ingest them through the same filter and envelope semantics the sinks publish with.

## What the dashboard page configures

Open **Developer → Event Sources** in the dashboard. The page ships one source kind today, `kafka`:

* **Enabled toggle** — when off, configuration is saved but nothing is consumed.
* **Topic** — the Kafka topic to consume from. 1–249 chars, `[A-Za-z0-9._-]`, never `.` or `..`.
* **Consumer group** — Kafka's `group.id` for this subscription. 1–255 chars, `[A-Za-z0-9._-]`.
* **Type path (optional)** — dot-path into each decoded message that carries the event type. Defaults to `type`; set `meta.event` if your envelope nests it.
* **Bootstrap brokers** — one `host:port` per line (no scheme, no path), up to 16.
* **SASL mechanism + username** — `plain`, `scram-sha-256`, or `scram-sha-512`. Skip SASL entirely if your brokers are not SASL-protected.
* **Event filter (optional)** — one pattern per line: an exact type (`order.created`), a prefix glob (`order.*`), or `*` / blank to ingest everything. The filter vocabulary is shared verbatim with event sinks, so inbound and outbound subscription semantics never drift.
* **Credentials** — the SASL password. Encrypted at rest, write-only: a read surfaces only a `credentials_configured` presence flag, so the password input always starts blank and is sent only when you type a new value. You can also permanently remove the stored secret with confirmation.

## API backing

Everything the page does is available under `/api/v1/developer/event-sources/:kind` where `:kind` is currently `kafka` only:

| Endpoint                                                   | Purpose                                                                                                                                                                                                                   |
| ---------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `GET /api/v1/developer/event-sources/kafka`                | Read the current config. The encrypted credential is never returned — only `credentials_configured: true \| false`.                                                                                                       |
| `PATCH /api/v1/developer/event-sources/kafka`              | Merge config — `enabled`, `source` (topic + consumer group + type path), `filter`, `brokers`, SASL identifiers, and/or a new `credentials` value. Empty-string `credentials` clears the stored secret.                    |
| `POST /api/v1/developer/event-sources/kafka/consume-batch` | Dry-run a sample batch of raw Kafka messages through the consume engine. Returns per-message dispositions and per-partition commit offsets. Nothing is persisted, no broker connection is opened, no consumer is started. |

A `PATCH` example:

```bash theme={null}
curl -X PATCH "https://api.orbit.devotel.io/api/v1/developer/event-sources/kafka" \
  -H "X-API-Key: dv_live_sk_your_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "enabled": true,
    "source": {
      "kind": "kafka",
      "topic": "customer.orders",
      "consumer_group": "orbit-inbound-consumer",
      "type_path": "type"
    },
    "filter": ["order.*", "payment.completed"],
    "brokers": ["broker-1.example.com:9092", "broker-2.example.com:9092"],
    "sasl_mechanism": "scram-sha-512",
    "sasl_username": "orbit-consumer",
    "credentials": "the-sasl-password"
  }'
```

The `source.kind` discriminator must equal the path segment — a `PATCH /…​/kafka` with `source.kind` of anything else returns `400`. Broker entries that carry a scheme or path (`https://…​:9092`) are rejected the same way.

## Test consume batch — dry-run walkthrough

Before any transport runs, you can validate your topic, type path, and filter against real sample messages. In the dashboard's **Test consume batch** panel, paste one raw JSON message per line:

```json theme={null}
{"type":"order.created","order_id":"order_123"}
{"type":"payment.completed","amount":99.0}
{"message":"heartbeat"}
{"type":"order.created","order_id":"order_123"}
```

Or call the endpoint directly:

```bash theme={null}
curl -X POST "https://api.orbit.devotel.io/api/v1/developer/event-sources/kafka/consume-batch" \
  -H "X-API-Key: dv_live_sk_your_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "topic": "customer.orders",
    "type_path": "type",
    "filter": ["order.*", "payment.completed"],
    "messages": [
      { "partition": 0, "offset": 42, "value": "{\"type\":\"order.created\",\"order_id\":\"order_123\"}" },
      { "partition": 0, "offset": 43, "value": "{\"type\":\"payment.completed\",\"amount\":99.0}" },
      { "partition": 0, "offset": 44, "value": "{\"message\":\"heartbeat\"}" },
      { "partition": 0, "offset": 42, "value": "{\"type\":\"order.created\",\"order_id\":\"order_123\"}", "key": "order_123" }
    ]
  }'
```

The response assigns every message exactly one terminal disposition:

```json theme={null}
{
  "data": {
    "kind": "kafka",
    "topic": "customer.orders",
    "received": 4,
    "landable": 2,
    "duplicate": 1,
    "filtered": 0,
    "dead_letter": 1,
    "commit_offsets": { "0": 44 },
    "persisted": false,
    "results": [
      { "partition": 0, "offset": 42, "disposition": "landable", "ingestId": "ing_…", "envelope": { "type": "order.created", "data": { "order_id": "order_123" } }, "errors": [], "warnings": [] },
      { "partition": 0, "offset": 43, "disposition": "landable", "ingestId": "ing_…", "envelope": { "type": "payment.completed", "data": { "amount": 99.0 } }, "errors": [], "warnings": [] },
      { "partition": 0, "offset": 44, "disposition": "dead_letter", "ingestId": "ing_…", "errors": [{ "code": "missing_type", "message": "A non-empty event type is required at 'type'." }], "warnings": [] },
      { "partition": 0, "offset": 42, "disposition": "duplicate", "key": "order_123", "ingestId": "ing_…", "errors": [], "warnings": [] }
    ]
  }
}
```

Read the dispositions as:

* **landable** — decoded, typed, and subscribed; the `envelope` carries the resolved event `type` plus the verbatim decoded `data`.
* **duplicate** — a redelivery of a coordinate `(topic, partition, offset)` already seen in this batch; skipped so you never double-ingest.
* **filtered** — well-formed but the resolved type is not in your filter; skipped deliberately.
* **dead\_letter** — undecodable (`invalid_json`), not a JSON object (`not_an_object`), or missing the event type at the configured path (`missing_type`). Routed aside with structured `errors` so a poison message never blocks the partition.

`commit_offsets` reports the highest safe-to-commit offset per partition — every message at or below it reached a terminal disposition, so advancing the offset never drops nor re-skips a record. `persisted: false` on the read is deliberate: the dry run never writes anything.

## Failure notes

* **Credentials** are encrypted at rest and never returned by any read — a `GET` surfaces only `credentials_configured`. Rotating means saving a new value; removing means an explicit empty-string clear.
* **The dry run is honest about its limits** — it does not persist envelopes and does not start a managed consumer. The long-running broker consumption transport is a tracked follow-up; what ships today is the config, validation, and the deterministic consume engine that transport will call per poll.
* **Feature-flagged transport** — the consume engine is inert until the transport flag is enabled by an operator, so flipping `enabled: true` on the config alone cannot start broker consumption.
* **Poison messages** dead-letter rather than stall a partition, and `commit_offsets` still advances past them.

## When inbound events are useful

Use an event source when Kafka is your system of record and Orbit should mirror a baseline stream of your domain events — order creation driving a follow-up campaign, a payment-completed signal closing out a billing interaction, CRM updates refreshing contact profiles. The inbound path shares the subscription-filter model with sinks, so choosing what to ingest is the same decision, inverted.

## Next steps

* [Event sinks streaming](/guides/event-sinks-streaming) — the outbound mirror; one event taxonomy, one filter vocabulary, both directions.
* [Event Sinks API](/api-reference/event-sinks) — the outbound sibling full field reference.
* [Webhook fan-out](/concepts/webhook-fan-out-and-event-sinks) — how sinks fit the push model the inbound source mirrors.