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

# Kafka Route

> Stream data both ways between your Coreflux broker and Apache Kafka clusters, including Confluent Cloud, AWS MSK, Aiven, and self-managed brokers

## Kafka Overview

The `KAFKA` route connects your Coreflux broker to an Apache Kafka cluster and moves messages in both directions. A single route can **produce** (MQTT → Kafka) and **consume** (Kafka → MQTT) at the same time, so device telemetry can flow into your data pipeline while commands and setpoints flow back to the field.

<Tip>
  **Like a conveyor belt between two warehouses.** Your MQTT broker handles fast-moving device traffic; Kafka is the long-term, replayable record. The Kafka route is the belt that carries boxes (messages) between them in whichever direction you point it.
</Tip>

### When to use it

* Feed real-time MQTT telemetry into a Kafka-based analytics or data pipeline.
* Bring Kafka events (commands, setpoints, alarms) back into the broker for field devices.
* Bridge edge MQTT traffic to a managed cloud platform like Confluent Cloud or AWS MSK.

### Key capabilities

| Capability              | What it means for you                                                                      |
| ----------------------- | ------------------------------------------------------------------------------------------ |
| **Bidirectional**       | Produce and consume in one route — no separate definitions needed                          |
| **Secure**              | Supports `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, and `SASL_SSL`                              |
| **Reliable delivery**   | Per-message offset commit on the consume path (at-least-once)                              |
| **Message keys**        | Route records to partitions by topic segment or JSON field for ordered, per-device streams |
| **Tunable throughput**  | Configurable batching, compression, and acknowledgement levels                             |
| **Cloud & self-hosted** | Confluent Cloud, AWS MSK, Aiven, Redpanda, and self-managed clusters                       |

***

Let's look at how a Kafka route is defined, starting with the three common directions.

## Quick Start

<Tabs>
  <Tab title="Produce (MQTT → Kafka)">
    Send every sensor reading from the broker into a Kafka topic, keyed by device:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE SensorProducer WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
            WITH GROUP_ID "coreflux-sensors"
            WITH SECURITY_PROTOCOL "PLAINTEXT"
        ADD EVENT SensorsToKafka
            WITH SOURCE_TOPIC "factory/sensors/+"
            WITH DESTINATION_TOPIC "factory.sensors.raw"
            WITH DIRECTION "out"
            WITH QUERY "{topic.3}"
    ```

    A message on `factory/sensors/pump-01` becomes a Kafka record on `factory.sensors.raw` with key `pump-01`.
  </Tab>

  <Tab title="Consume (Kafka → MQTT)">
    Forward Kafka commands into the MQTT broker:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE CommandConsumer WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
            WITH GROUP_ID "coreflux-commands"
            WITH SECURITY_PROTOCOL "PLAINTEXT"
            WITH AUTO_OFFSET_RESET "Latest"
        ADD EVENT CommandsFromKafka
            WITH SOURCE_TOPIC "factory.commands"
            WITH DESTINATION_TOPIC "factory/commands/in"
            WITH DIRECTION "in"
    ```

    A Kafka record on `factory.commands` is published to the MQTT topic `factory/commands/in`.
  </Tab>

  <Tab title="Bidirectional">
    One route handling both directions — telemetry out, commands in:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE PlantKafka WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
            WITH GROUP_ID "coreflux-plant"
            WITH SECURITY_PROTOCOL "PLAINTEXT"
            WITH AUTO_OFFSET_RESET "Earliest"
        ADD EVENT SensorsToKafka
            WITH SOURCE_TOPIC "plant/sensors/+"
            WITH DESTINATION_TOPIC "plant.sensors.raw"
            WITH DIRECTION "out"
        ADD EVENT CommandsFromKafka
            WITH SOURCE_TOPIC "plant.commands"
            WITH DESTINATION_TOPIC "plant/commands/in"
            WITH DIRECTION "in"
    ```
  </Tab>
</Tabs>

<Warning>
  The Quick Start examples use `PLAINTEXT` for clarity. For any cluster reachable over a network — especially cloud services — use `SASL_SSL` and load credentials from environment variables and secrets. See [Security & Authentication](#security--authentication).
</Warning>

***

## How Kafka Compares to MQTT

Kafka and MQTT solve different problems. Coreflux bridges the two so each does what it's best at: MQTT for fast device traffic, Kafka for durable, replayable streams.

| Feature                      | Kafka                             | MQTT                          |
| ---------------------------- | --------------------------------- | ----------------------------- |
| **Delivery model**           | Pull (consumer fetches)           | Push (broker delivers)        |
| **Retention**                | Configurable (hours to forever)   | Typically none                |
| **Ordering**                 | Per-partition                     | Per-topic                     |
| **Fan-out / load balancing** | Built-in consumer groups          | Shared subscriptions          |
| **Throughput**               | Millions of messages/sec          | Thousands of messages/sec     |
| **Typical use**              | Data pipelines, analytics, replay | IoT telemetry, device control |

A few Kafka terms appear throughout this page:

| Term                  | Description                                                                         |
| --------------------- | ----------------------------------------------------------------------------------- |
| **Bootstrap servers** | The initial broker addresses the route contacts to discover the rest of the cluster |
| **Topic**             | A named, partitioned log of records                                                 |
| **Partition**         | An ordered sub-stream within a topic — the unit of ordering and parallelism         |
| **Consumer group**    | A set of consumers that divide a topic's partitions between them                    |
| **Offset**            | A record's position within a partition, committed once it has been processed        |

***

## Supported Kafka Services

The Kafka route works with any Kafka-compatible cluster, cloud-managed or self-hosted.

<Tabs>
  <Tab title="Cloud-managed">
    | Service                          | Security Protocol | Notes                                      |
    | -------------------------------- | ----------------- | ------------------------------------------ |
    | **Confluent Cloud**              | `SASL_SSL`        | `PLAIN` or `SCRAM` mechanism               |
    | **Amazon MSK**                   | `SASL_SSL`        | IAM or SCRAM auth                          |
    | **Aiven for Kafka**              | `SASL_SSL`        | SCRAM-SHA-512 typical                      |
    | **Azure Event Hubs (Kafka API)** | `SASL_SSL`        | `PLAIN` with connection string as password |
    | **Redpanda Cloud**               | `SASL_SSL`        | Kafka-compatible                           |
  </Tab>

  <Tab title="Self-managed">
    | Cluster                 | Notes                                     |
    | ----------------------- | ----------------------------------------- |
    | **Apache Kafka** (2.0+) | All security protocols supported          |
    | **Confluent Platform**  | Same configuration as Apache Kafka        |
    | **Redpanda**            | Drop-in Kafka-compatible cluster          |
    | **KRaft mode**          | Kafka without ZooKeeper — fully supported |
  </Tab>
</Tabs>

***

## Connection Configuration

Every Kafka route starts with a `KAFKA_CONFIG` block describing the cluster, security, and throughput settings. Only `BROKER_ADDRESS` is required, but production routes should always set a `GROUP_ID` and use a secure protocol.

```lot wrap focus={3-4} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE PlantKafka WITH TYPE KAFKA
    ADD KAFKA_CONFIG
        WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
        WITH GROUP_ID "coreflux-plant"
    ADD EVENT SensorsToKafka
        WITH SOURCE_TOPIC "plant/sensors/+"
        WITH DESTINATION_TOPIC "plant.sensors.raw"
        WITH DIRECTION "out"
```

### Connection & Security

<AccordionGroup>
  <Accordion title="Cluster Connection">
    <ParamField path="BROKER_ADDRESS" type="string" required>
      Bootstrap servers, e.g. `localhost:9092` or `host1:9092,host2:9092`. List two or three brokers in production so the route survives an individual broker restart.
    </ParamField>

    <ParamField path="GROUP_ID" type="string">
      Consumer group name. **Always set this in production.** If omitted, a random group is generated on every startup, causing the consumer to replay all retained messages from scratch after each restart.
    </ParamField>

    <ParamField path="AUTO_OFFSET_RESET" type="string" default="Earliest">
      Where to start reading when no committed offset exists: `Earliest` (from the beginning) or `Latest` (new messages only).
    </ParamField>

    <ParamField path="ENABLE_AUTO_COMMIT" type="boolean" default="false">
      Leave `false` for per-message commit (at-least-once delivery). Set `true` to additionally enable periodic background commits.
    </ParamField>
  </Accordion>

  <Accordion title="Security">
    <ParamField path="SECURITY_PROTOCOL" type="string" default="PLAINTEXT">
      `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, or `SASL_SSL`.
    </ParamField>

    <ParamField path="SASL_MECHANISM" type="string">
      Required with SASL: `PLAIN`, `SCRAM-SHA-256`, or `SCRAM-SHA-512`.
    </ParamField>

    <ParamField path="SASL_USERNAME" type="string">
      SASL username. Supports `GET ENV`.
    </ParamField>

    <ParamField path="SASL_PASSWORD" type="string">
      SASL password. Use `GET SECRET` to keep it out of your LoT definition.
    </ParamField>

    <ParamField path="SSL_CA_LOCATION" type="string">
      Path to the CA certificate (for `SSL` or `SASL_SSL`).
    </ParamField>

    <ParamField path="SSL_CERTIFICATE_LOCATION" type="string">
      Path to the client certificate (mutual TLS).
    </ParamField>

    <ParamField path="SSL_KEY_LOCATION" type="string">
      Path to the client key (required with a client certificate).
    </ParamField>

    <ParamField path="SSL_KEYSTORE_PASSWORD" type="string">
      Key or keystore password.
    </ParamField>
  </Accordion>

  <Accordion title="Throughput & Tuning">
    <ParamField path="ACKS" type="string" default="all">
      Producer durability: `0` (no acknowledgement), `1` (leader only), or `all` (all in-sync replicas).
    </ParamField>

    <ParamField path="LINGER_MS" type="integer" default="5">
      How long the producer waits to fill a batch before sending. Higher values mean fewer, larger requests (better throughput) at the cost of a little added latency.
    </ParamField>

    <ParamField path="BATCH_SIZE" type="integer">
      Maximum producer batch size in bytes. Default is 1 MB.
    </ParamField>

    <ParamField path="COMPRESSION_TYPE" type="string" default="none">
      Producer compression: `none`, `gzip`, `snappy`, `lz4`, or `zstd`. Most effective on larger batches — pair with `LINGER_MS`.
    </ParamField>

    <ParamField path="FETCH_MAX_BYTES" type="integer">
      Maximum bytes returned per consumer fetch. Default is 50 MB.
    </ParamField>

    <ParamField path="QUEUED_MAX_MESSAGES_KB" type="integer" default="65536">
      Maximum KB pre-fetched per partition into the consumer buffer (64 MB by default). Lower it to cap memory use on high-volume topics.
    </ParamField>

    <ParamField path="FLUSH_TIMEOUT_MS" type="integer" default="5000">
      How long the route waits for in-flight produced messages to drain when stopping.
    </ParamField>
  </Accordion>
</AccordionGroup>

***

## Security & Authentication

Choose the security protocol that matches your broker's listener. For anything beyond a local development cluster, use `SASL_SSL`.

| `SECURITY_PROTOCOL` | TLS | SASL | Typical use                             |
| ------------------- | --- | ---- | --------------------------------------- |
| `PLAINTEXT`         | No  | No   | Local development                       |
| `SSL`               | Yes | No   | TLS-only clusters                       |
| `SASL_PLAINTEXT`    | No  | Yes  | Internal test environments              |
| `SASL_SSL`          | Yes | Yes  | Confluent Cloud, MSK, Aiven, production |

<Tabs>
  <Tab title="PLAINTEXT (development)">
    No encryption or authentication — local clusters only:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE LocalKafka WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS "localhost:9092"
            WITH GROUP_ID "coreflux-dev"
            WITH SECURITY_PROTOCOL "PLAINTEXT"
        ADD EVENT SensorsToKafka
            WITH SOURCE_TOPIC "dev/sensors/+"
            WITH DESTINATION_TOPIC "dev.sensors"
            WITH DIRECTION "out"
    ```
  </Tab>

  <Tab title="SSL (TLS only)">
    Encrypted transport without SASL credentials:

    ```lot wrap focus={5-6} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE TlsKafka WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS "kafka.example.com:9093"
            WITH GROUP_ID "coreflux-tls"
            WITH SECURITY_PROTOCOL "SSL"
            WITH SSL_CA_LOCATION "/etc/ssl/certs/kafka-ca.pem"
        ADD EVENT SensorsToKafka
            WITH SOURCE_TOPIC "plant/sensors/+"
            WITH DESTINATION_TOPIC "plant.sensors"
            WITH DIRECTION "out"
    ```
  </Tab>

  <Tab title="SASL_SSL (production)">
    Encrypted transport with SCRAM authentication — the recommended production setup:

    ```lot wrap focus={5-10} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE SecureKafka WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
            WITH GROUP_ID "coreflux-prod"
            WITH SECURITY_PROTOCOL "SASL_SSL"
            WITH SASL_MECHANISM "SCRAM-SHA-512"
            WITH SASL_USERNAME GET ENV "KAFKA_USER"
            WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"
            WITH SSL_CA_LOCATION "/etc/ssl/certs/ca.pem"
        ADD EVENT SensorsToKafka
            WITH SOURCE_TOPIC "plant/sensors/+"
            WITH DESTINATION_TOPIC "plant.sensors"
            WITH DIRECTION "out"
    ```
  </Tab>
</Tabs>

<Note>
  Store credentials with `GET SECRET` and non-sensitive values like hostnames with `GET ENV`. See [Environment Variables and Secrets](/latest/mqtt-broker/secrets-and-env) for setup across Windows, Linux, Docker, and Kubernetes.
</Note>

***

## Events and Direction

Each `ADD EVENT` block maps one MQTT topic to one Kafka topic and sets the direction of flow. Think of the **broker** as the source side and **Kafka** as the destination side.

| `DIRECTION` | `SOURCE_TOPIC`                   | `DESTINATION_TOPIC`             | Data flow    |
| ----------- | -------------------------------- | ------------------------------- | ------------ |
| `"out"`     | Broker MQTT topic (wildcards OK) | Kafka topic to produce to       | MQTT → Kafka |
| `"in"`      | Kafka topic to subscribe to      | Broker MQTT topic to publish to | Kafka → MQTT |

### Event Parameters

<AccordionGroup>
  <Accordion title="Required">
    <ParamField path="SOURCE_TOPIC" type="string" required>
      Broker MQTT topic when `DIRECTION` is `"out"`; Kafka topic to subscribe to when `DIRECTION` is `"in"`.
    </ParamField>

    <ParamField path="DESTINATION_TOPIC" type="string" required>
      Kafka topic when `DIRECTION` is `"out"`; broker MQTT publish topic when `DIRECTION` is `"in"`.
    </ParamField>

    <ParamField path="DIRECTION" type="string" required>
      `"out"` (MQTT → Kafka) or `"in"` (Kafka → MQTT).
    </ParamField>
  </Accordion>

  <Accordion title="Optional">
    <ParamField path="QUERY" type="string">
      The Kafka record **key** for partitioning (produce events only). See [Message Keys and Partitioning](#message-keys-and-partitioning).
    </ParamField>

    <ParamField path="EVERY" type="interval">
      Produce on a timer instead of on each incoming message.
    </ParamField>
  </Accordion>
</AccordionGroup>

***

## Message Keys and Partitioning

Kafka sends records with the same key to the same partition, which preserves **order per key**. On produce events, set the key with `WITH QUERY`.

* **With a key** (`WITH QUERY`): same key → same partition → ordering preserved per key (for example, all readings from one device stay in order).
* **Without a key**: records spread across all partitions for higher throughput, with no per-key ordering guarantee.
* **Consume events** join the consumer group and have partitions assigned automatically — set a stable `GROUP_ID` so offsets survive restarts.

### Key Placeholders

The `WITH QUERY` value is resolved before each record is sent. Supported placeholders:

| Placeholder            | Description                               | Example result                                      |
| ---------------------- | ----------------------------------------- | --------------------------------------------------- |
| `{topic.N}`            | Nth segment (1-based) of the source topic | `pump-01` from `plant/sensors/pump-01/reading`      |
| `{topic}`              | Full source topic string                  | `plant/sensors/pump-01/reading`                     |
| `{payload}`            | Message payload as a string               | `42.5`                                              |
| `{payload.json}`       | Full payload parsed as JSON               | —                                                   |
| `{payload.json.field}` | A JSON field from the payload             | `pump-01` from `{"device_id":"pump-01","value":42}` |
| `{timestamp}`          | ISO 8601 timestamp                        | `2026-06-12T10:30:45.123Z`                          |
| `{env.NAME}`           | Environment variable value                | value of `NAME`                                     |
| `{secret.NAME}`        | Secret value                              | value of `NAME`                                     |

### Key from a topic segment

Use a wildcard position from the source topic as the key — here the device ID in the third segment:

```lot wrap focus={10} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE SensorsByDevice WITH TYPE KAFKA
    ADD KAFKA_CONFIG
        WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
        WITH GROUP_ID "coreflux-sensors"
        WITH SECURITY_PROTOCOL "SASL_SSL"
        WITH SASL_MECHANISM "SCRAM-SHA-512"
        WITH SASL_USERNAME GET ENV "KAFKA_USER"
        WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"
    ADD EVENT SensorsByDevice
        WITH SOURCE_TOPIC "plant/sensors/+/reading"
        WITH DESTINATION_TOPIC "plant.sensors.raw"
        WITH DIRECTION "out"
        WITH QUERY "{topic.3}"
```

A message on `plant/sensors/pump-01/reading` gets the key `pump-01`.

### Key from a JSON field

When the identifier lives inside the payload, pull it from a JSON field instead:

```lot wrap focus={13} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE SensorsByPayloadId WITH TYPE KAFKA
    ADD KAFKA_CONFIG
        WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
        WITH GROUP_ID "coreflux-sensors"
        WITH SECURITY_PROTOCOL "SASL_SSL"
        WITH SASL_MECHANISM "SCRAM-SHA-512"
        WITH SASL_USERNAME GET ENV "KAFKA_USER"
        WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"
    ADD EVENT SensorsByPayloadId
        WITH SOURCE_TOPIC "plant/sensors/+/reading"
        WITH DESTINATION_TOPIC "plant.sensors.raw"
        WITH DIRECTION "out"
        WITH QUERY "{payload.json.device_id}"
```

For a payload of `{"device_id":"pump-01","value":42}`, the record key becomes `pump-01`.

***

## Complete Examples

<Tabs>
  <Tab title="Production (SASL_SSL)">
    Full production route with TLS, SCRAM authentication, telemetry out, and alarms in:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE ProductionKafka WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
            WITH GROUP_ID "coreflux-production"
            WITH SECURITY_PROTOCOL "SASL_SSL"
            WITH SASL_MECHANISM "SCRAM-SHA-512"
            WITH SASL_USERNAME GET ENV "KAFKA_USER"
            WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"
            WITH SSL_CA_LOCATION "/etc/ssl/certs/ca.pem"
            WITH ACKS "all"
            WITH AUTO_OFFSET_RESET "Earliest"
        ADD EVENT TelemetryOut
            WITH SOURCE_TOPIC "production/+/telemetry"
            WITH DESTINATION_TOPIC "production.telemetry"
            WITH DIRECTION "out"
            WITH QUERY "{topic.2}"
        ADD EVENT AlarmsIn
            WITH SOURCE_TOPIC "production.alarms"
            WITH DESTINATION_TOPIC "production/alarms/incoming"
            WITH DIRECTION "in"
    ```
  </Tab>

  <Tab title="High-throughput telemetry">
    Optimized for sustained high-rate telemetry with batching and compression:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE TelemetryKafka WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
            WITH GROUP_ID "coreflux-telemetry"
            WITH SECURITY_PROTOCOL "SASL_SSL"
            WITH SASL_MECHANISM "SCRAM-SHA-512"
            WITH SASL_USERNAME GET ENV "KAFKA_USER"
            WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"
            WITH ACKS "1"
            WITH LINGER_MS "50"
            WITH COMPRESSION_TYPE "lz4"
            WITH FLUSH_TIMEOUT_MS "10000"
        ADD EVENT DeviceTelemetry
            WITH SOURCE_TOPIC "devices/+/telemetry"
            WITH DESTINATION_TOPIC "iot.telemetry"
            WITH DIRECTION "out"
            WITH QUERY "{topic.2}"
    ```

    `ACKS "1"` avoids waiting for every replica on non-critical telemetry, `LINGER_MS "50"` batches bursts into fewer requests, and `lz4` compression cuts bandwidth with negligible CPU cost.
  </Tab>

  <Tab title="Multiple topic mappings">
    Route different asset types to different Kafka topics, keyed by a JSON field:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE AssetKafka WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS GET ENV "KAFKA_BOOTSTRAP"
            WITH GROUP_ID "coreflux-assets"
            WITH SECURITY_PROTOCOL "SASL_SSL"
            WITH SASL_MECHANISM "SCRAM-SHA-512"
            WITH SASL_USERNAME GET ENV "KAFKA_USER"
            WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"
        ADD EVENT PumpsOut
            WITH SOURCE_TOPIC "assets/pumps/+/data"
            WITH DESTINATION_TOPIC "assets.pumps"
            WITH DIRECTION "out"
            WITH QUERY "{payload.json.asset_id}"
        ADD EVENT MotorsOut
            WITH SOURCE_TOPIC "assets/motors/+/data"
            WITH DESTINATION_TOPIC "assets.motors"
            WITH DIRECTION "out"
            WITH QUERY "{payload.json.asset_id}"
        ADD EVENT SetpointsIn
            WITH SOURCE_TOPIC "assets.setpoints"
            WITH DESTINATION_TOPIC "assets/setpoints/incoming"
            WITH DIRECTION "in"
    ```
  </Tab>

  <Tab title="Confluent Cloud">
    Connect to Confluent Cloud using SASL/PLAIN with an API key and secret:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE ConfluentCloud WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS GET ENV "CONFLUENT_BOOTSTRAP"
            WITH GROUP_ID "coreflux-cloud"
            WITH SECURITY_PROTOCOL "SASL_SSL"
            WITH SASL_MECHANISM "PLAIN"
            WITH SASL_USERNAME GET ENV "CONFLUENT_API_KEY"
            WITH SASL_PASSWORD GET SECRET "CONFLUENT_API_SECRET"
            WITH ACKS "all"
            WITH AUTO_OFFSET_RESET "Latest"
        ADD EVENT TelemetryToCloud
            WITH SOURCE_TOPIC "factory/+/telemetry"
            WITH DESTINATION_TOPIC "factory.telemetry"
            WITH DIRECTION "out"
            WITH QUERY "{topic.2}"
    ```

    `CONFLUENT_BOOTSTRAP` is the `pkc-xxxx.region.confluent.cloud:9092` address from your cluster settings; the API key and secret come from the Confluent Cloud credentials page.
  </Tab>

  <Tab title="High availability">
    List multiple bootstrap brokers so the route survives a single broker restart:

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    DEFINE ROUTE HaKafka WITH TYPE KAFKA
        ADD KAFKA_CONFIG
            WITH BROKER_ADDRESS "kafka-1:9092,kafka-2:9092,kafka-3:9092"
            WITH GROUP_ID "coreflux-ha"
            WITH SECURITY_PROTOCOL "SASL_SSL"
            WITH SASL_MECHANISM "SCRAM-SHA-512"
            WITH SASL_USERNAME GET ENV "KAFKA_USER"
            WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"
            WITH ACKS "all"
        ADD EVENT TelemetryOut
            WITH SOURCE_TOPIC "factory/sensors/+"
            WITH DESTINATION_TOPIC "factory.sensors"
            WITH DIRECTION "out"
    ```
  </Tab>
</Tabs>

***

## Performance Tuning

A few settings control the trade-off between throughput, latency, durability, and memory. Tune them to match each stream.

### Durability vs. throughput (produce)

`ACKS` decides how many replicas must confirm each record before it counts as sent:

| `ACKS` | Behavior                                                       | Use for                                          |
| ------ | -------------------------------------------------------------- | ------------------------------------------------ |
| `all`  | Wait for all in-sync replicas (most durable, lower throughput) | Mission-critical data                            |
| `1`    | Wait for the partition leader only                             | High-rate telemetry                              |
| `0`    | Don't wait at all                                              | Fire-and-forget metrics where loss is acceptable |

### Batching and compression (produce)

Under burst load, many produces are in flight at once and Kafka groups them into shared batches. These settings shape that behavior:

| Setting                  | Effect                                                    |
| ------------------------ | --------------------------------------------------------- |
| `LINGER_MS "0"`          | Send immediately — lowest latency, smallest batches       |
| `LINGER_MS "50"`         | Wait up to 50 ms to fill a batch — fewer, larger requests |
| `COMPRESSION_TYPE "lz4"` | Compress batches — lower bandwidth at negligible CPU cost |
| `BATCH_SIZE "2097152"`   | Allow 2 MB batches — more messages per request            |

### Consume throughput

Each consumed message is committed individually for at-least-once delivery, which adds a short round-trip per message. For very high-rate consume topics, use one Kafka route per logical stream rather than stacking many consume events on a single route.

### Memory use (consume)

The route pre-fetches messages into a buffer while it processes. On high-volume topics with many partitions this buffer can grow large. Cap it with `QUEUED_MAX_MESSAGES_KB` (for example `"4096"` for 4 MB per partition) and bound fetch responses with `FETCH_MAX_BYTES`.

### Watch for consume backpressure

When a consume route publishes into the broker, a busy MQTT subsystem (many subscribers, large payloads) can slow the consumer down. If you pair a high-rate consume topic with many subscribers, monitor consumer group lag to catch this early.

***

## Best Practices

<AccordionGroup>
  <Accordion title="Always set a stable GROUP_ID">
    A fixed group ID lets offsets survive restarts. Without one, a new random group is created on every startup and the consumer replays the entire topic.

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    WITH GROUP_ID "coreflux-plant-v1"
    ```
  </Accordion>

  <Accordion title="Use GET SECRET for credentials">
    Never hardcode passwords in a route definition.

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"
    ```
  </Accordion>

  <Accordion title="Use SASL_SSL in production">
    Reserve `PLAINTEXT` for local development; encrypt and authenticate everything else.

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    WITH SECURITY_PROTOCOL "SASL_SSL"
    WITH SASL_MECHANISM "SCRAM-SHA-512"
    ```
  </Accordion>

  <Accordion title="Set AUTO_OFFSET_RESET deliberately">
    Choose where a brand-new consumer starts reading.

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    WITH AUTO_OFFSET_RESET "Earliest"   // replay history
    WITH AUTO_OFFSET_RESET "Latest"     // new messages only
    ```
  </Accordion>

  <Accordion title="Key messages for per-device ordering">
    Use a message key so all records from one device land on the same partition, in order.

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    WITH QUERY "{topic.2}"
    ```
  </Accordion>

  <Accordion title="List multiple bootstrap brokers">
    A single broker address is a single point of failure; list several for resilience.

    ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
    WITH BROKER_ADDRESS "kafka-1:9092,kafka-2:9092,kafka-3:9092"
    ```
  </Accordion>

  <Accordion title="Monitor consumer group lag">
    Use your Kafka management UI (such as Confluent Control Center or Redpanda Console) or `kafka-consumer-groups --describe --group <GROUP_ID>` to watch for growing lag, which signals publish backpressure.
  </Accordion>
</AccordionGroup>

***

## Troubleshooting

<AccordionGroup>
  <Accordion title="Quick connection check">
    Verify a route's connection at any time by publishing the following to `$SYS/Coreflux/Command` from any MQTT client:

    ```
    -checkRouteConnection PlantKafka
    ```

    The route validates the cluster connection and, when consume events are present, the consumer group join — catching configuration problems early.
  </Accordion>

  <Accordion title="Cannot connect to the cluster">
    * Confirm the bootstrap address and port are correct.
    * Make sure `SECURITY_PROTOCOL` matches the broker's listener.
    * Check that the port (default `9092`) isn't blocked by a firewall.
    * Verify network reachability to the broker host.
  </Accordion>

  <Accordion title="Connects, but no messages flow">
    The bootstrap connection can succeed while the cluster's advertised addresses point somewhere unreachable. This is common when Kafka runs in Docker. Make sure the broker advertises an address the Coreflux host can actually reach:

    | Where Coreflux runs          | `BROKER_ADDRESS` |
    | ---------------------------- | ---------------- |
    | On the host                  | `localhost:9092` |
    | Same Docker network as Kafka | `kafka:9092`     |
  </Accordion>

  <Accordion title="Authentication or SASL errors">
    * Confirm `SECURITY_PROTOCOL` matches the listener (e.g. `SASL_SSL`, not `SASL_PLAINTEXT`).
    * Make sure `SASL_MECHANISM` matches what the broker requires.
    * Check that credentials are correct and the env/secret names resolve.
  </Accordion>

  <Accordion title="Consumer receives nothing / replays every restart">
    * Set `AUTO_OFFSET_RESET "Earliest"` if messages were produced before the consumer first started.
    * Set a stable `GROUP_ID` so offsets are remembered across restarts.
    * Give each route its own `GROUP_ID` unless you intentionally want them to share partitions.
  </Accordion>

  <Accordion title="Performance problems">
    | Symptom                       | Likely cause                       | Fix                                               |
    | ----------------------------- | ---------------------------------- | ------------------------------------------------- |
    | Produce slower than expected  | `ACKS "all"` on a high-rate stream | Use `ACKS "1"` for telemetry                      |
    | High bandwidth usage          | No compression                     | Add `COMPRESSION_TYPE "lz4"` and `LINGER_MS "50"` |
    | High memory on consume routes | Large prefetch buffer              | Set `QUEUED_MAX_MESSAGES_KB "4096"`               |
    | Growing consumer lag          | MQTT publish backpressure          | Reduce subscriber load on the destination topics  |
    | Messages lost on shutdown     | Flush timeout too short            | Raise `FLUSH_TIMEOUT_MS`                          |
  </Accordion>
</AccordionGroup>

***

## Next Steps

<CardGroup cols={2}>
  <Card title="MQTT Bridge" icon="arrow-right-arrow-left" href="./mqtt-bridge">
    Sync topics between two MQTT brokers.
  </Card>

  <Card title="Data Storage Routes" icon="database" href="../data-storage/overview">
    Persist MQTT data in databases for analytics.
  </Card>
</CardGroup>