​

Configure event sources: Kafka inbound consume

​

What the dashboard page configures

• 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

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"
  }'

​

Test consume batch — dry-run walkthrough

{"type":"order.created","order_id":"order_123"}
{"type":"payment.completed","amount":99.0}
{"message":"heartbeat"}
{"type":"order.created","order_id":"order_123"}

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" }
    ]
  }'

{
  "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": [] }
    ]
  }
}

• 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.

​

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

​

Next steps

• Event sinks streaming — the outbound mirror; one event taxonomy, one filter vocabulary, both directions.
• Event Sinks API — the outbound sibling full field reference.
• Webhook fan-out — how sinks fit the push model the inbound source mirrors.