Kafka Source

The Kafka source subscribes to one or more Apache Kafka topics using a consumer group, making it easy to build event-driven pipelines that react to messages in real time.

If brokers is not set in the source block, the engine falls back to the root-level kafka configuration. This lets you define broker addresses once and share them across multiple channels.

Configuration

source:
  type: kafka
  brokers:
    - kafka-1.internal:9093
    - kafka-2.internal:9093
    - kafka-3.internal:9093
  topic: ehr.adt.events
  group_id: intuware-adt-consumer
  offset: earliest
  auth:
    mechanism: SASL_SSL
    username: ${KAFKA_USER}
    password: ${KAFKA_PASSWORD}
  tls:
    ca_file: ./certs/kafka-ca.crt

Properties

Property	Type	Required	Default	Description
`brokers`	`string[]`	No	—	List of Kafka broker addresses. Falls back to the root-level `kafka.brokers` if omitted.
`topic`	`string`	Yes	—	Kafka topic to subscribe to.
`group_id`	`string`	No	—	Consumer group ID. When set, Kafka tracks offsets and distributes partitions across group members.
`offset`	`string`	No	`latest`	Starting offset when no committed offset exists: `earliest` or `latest`.
`auth`	`object`	No	—	SASL authentication configuration.
`tls`	`object`	No	—	TLS settings for encrypted broker connections.

Authentication

Set auth.mechanism to one of the supported SASL mechanisms:

Mechanism	Description
`SASL_PLAIN`	Plain username / password (use with TLS).
`SCRAM-SHA-256`	Challenge-response with SHA-256.
`SCRAM-SHA-512`	Challenge-response with SHA-512.
`SASL_SSL`	SASL authentication over a TLS connection.

auth:
  mechanism: SCRAM-SHA-512
  username: ${KAFKA_USER}
  password: ${KAFKA_PASSWORD}

Complete Example

Consume ADT events from Kafka and push them to a FHIR server:

channel:
  name: kafka-adt-to-fhir
  description: Stream ADT events from Kafka into the FHIR repository

kafka:
  brokers:
    - kafka-1.internal:9093
    - kafka-2.internal:9093
    - kafka-3.internal:9093

source:
  type: kafka
  topic: ehr.adt.events
  group_id: intuware-adt-consumer
  offset: earliest
  auth:
    mechanism: SASL_SSL
    username: ${KAFKA_USER}
    password: ${KAFKA_PASSWORD}
  tls:
    ca_file: ./certs/kafka-ca.crt

transformer:
  type: typescript
  file: ./transforms/adt-to-fhir.ts

destination:
  type: http
  url: https://fhir.hospital.org/r4/Bundle
  method: POST
  headers:
    Content-Type: application/fhir+json