Kafka Route

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.

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.

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

Produce (MQTT → Kafka)
Consume (Kafka → MQTT)
Bidirectional

Send every sensor reading from the broker into a Kafka topic, keyed by device:

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.

Forward Kafka commands into the MQTT broker:

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.

One route handling both directions — telemetry out, commands in:

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"

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.

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.

Cloud-managed
Self-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

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

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.

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

Cluster Connection

BROKER_ADDRESS

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.

GROUP_ID

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.

AUTO_OFFSET_RESET

string

default:"Earliest"

Where to start reading when no committed offset exists: Earliest (from the beginning) or Latest (new messages only).

ENABLE_AUTO_COMMIT

boolean

default:"false"

Leave false for per-message commit (at-least-once delivery). Set true to additionally enable periodic background commits.

Security

SECURITY_PROTOCOL

string

default:"PLAINTEXT"

PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL.

SASL_MECHANISM

string

Required with SASL: PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512.

SASL_USERNAME

string

SASL username. Supports GET ENV.

SASL_PASSWORD

string

SASL password. Use GET SECRET to keep it out of your LoT definition.

SSL_CA_LOCATION

string

Path to the CA certificate (for SSL or SASL_SSL).

SSL_CERTIFICATE_LOCATION

string

Path to the client certificate (mutual TLS).

SSL_KEY_LOCATION

string

Path to the client key (required with a client certificate).

SSL_KEYSTORE_PASSWORD

string

Key or keystore password.

Throughput & Tuning

ACKS

string

default:"all"

Producer durability: 0 (no acknowledgement), 1 (leader only), or all (all in-sync replicas).

LINGER_MS

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.

BATCH_SIZE

integer

Maximum producer batch size in bytes. Default is 1 MB.

COMPRESSION_TYPE

string

default:"none"

Producer compression: none, gzip, snappy, lz4, or zstd. Most effective on larger batches — pair with LINGER_MS.

FETCH_MAX_BYTES

integer

Maximum bytes returned per consumer fetch. Default is 50 MB.

QUEUED_MAX_MESSAGES_KB

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.

FLUSH_TIMEOUT_MS

integer

default:"5000"

How long the route waits for in-flight produced messages to drain when stopping.

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

PLAINTEXT (development)
SSL (TLS only)
SASL_SSL (production)

No encryption or authentication — local clusters only:

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"

Encrypted transport without SASL credentials:

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"

Encrypted transport with SCRAM authentication — the recommended production setup:

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"

Store credentials with GET SECRET and non-sensitive values like hostnames with GET ENV. See Environment Variables and Secrets for setup across Windows, Linux, Docker, and Kubernetes.

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

Required

SOURCE_TOPIC

string

required

Broker MQTT topic when DIRECTION is "out"; Kafka topic to subscribe to when DIRECTION is "in".

DESTINATION_TOPIC

string

required

Kafka topic when DIRECTION is "out"; broker MQTT publish topic when DIRECTION is "in".

DIRECTION

string

required

"out" (MQTT → Kafka) or "in" (Kafka → MQTT).

Optional

QUERY

string

The Kafka record key for partitioning (produce events only). See Message Keys and Partitioning.

EVERY

interval

Produce on a timer instead of on each incoming message.

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:

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:

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

Production (SASL_SSL)
High-throughput telemetry
Multiple topic mappings
Confluent Cloud
High availability

Full production route with TLS, SCRAM authentication, telemetry out, and alarms in:

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"

Optimized for sustained high-rate telemetry with batching and compression:

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.

Route different asset types to different Kafka topics, keyed by a JSON field:

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"

Connect to Confluent Cloud using SASL/PLAIN with an API key and secret:

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.

List multiple bootstrap brokers so the route survives a single broker restart:

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"

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

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.

WITH GROUP_ID "coreflux-plant-v1"

Use GET SECRET for credentials

Never hardcode passwords in a route definition.

WITH SASL_PASSWORD GET SECRET "KAFKA_PASSWORD"

Use SASL_SSL in production

Reserve PLAINTEXT for local development; encrypt and authenticate everything else.

WITH SECURITY_PROTOCOL "SASL_SSL"
WITH SASL_MECHANISM "SCRAM-SHA-512"

Set AUTO_OFFSET_RESET deliberately

Choose where a brand-new consumer starts reading.

WITH AUTO_OFFSET_RESET "Earliest"   // replay history
WITH AUTO_OFFSET_RESET "Latest"     // new messages only

Key messages for per-device ordering

Use a message key so all records from one device land on the same partition, in order.

WITH QUERY "{topic.2}"

List multiple bootstrap brokers

A single broker address is a single point of failure; list several for resilience.

WITH BROKER_ADDRESS "kafka-1:9092,kafka-2:9092,kafka-3:9092"

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.

Troubleshooting

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.

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.

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`

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.

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.

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`

Kafka Overview

When to use it

Key capabilities

Quick Start

How Kafka Compares to MQTT

Supported Kafka Services

Connection Configuration

Connection & Security

Security & Authentication

Events and Direction

Event Parameters

Message Keys and Partitioning

Key Placeholders

Key from a topic segment

Key from a JSON field

Complete Examples

Performance Tuning

Durability vs. throughput (produce)

Batching and compression (produce)

Consume throughput

Memory use (consume)

Watch for consume backpressure

Best Practices

Troubleshooting

Next Steps

MQTT Bridge

Data Storage Routes

​Kafka Overview

​When to use it

​Key capabilities

​Quick Start

​How Kafka Compares to MQTT

​Supported Kafka Services

​Connection Configuration

​Connection & Security

​Security & Authentication

​Events and Direction

​Event Parameters

​Message Keys and Partitioning

​Key Placeholders

​Key from a topic segment

​Key from a JSON field

​Complete Examples

​Performance Tuning

​Durability vs. throughput (produce)

​Batching and compression (produce)

​Consume throughput

​Memory use (consume)

​Watch for consume backpressure

​Best Practices

​Troubleshooting

​Next Steps

MQTT Bridge

Data Storage Routes

Kafka Overview

When to use it

Key capabilities

Quick Start

How Kafka Compares to MQTT

Supported Kafka Services

Connection Configuration

Connection & Security

Security & Authentication

Events and Direction

Event Parameters

Message Keys and Partitioning

Key Placeholders

Key from a topic segment

Key from a JSON field

Complete Examples

Performance Tuning

Durability vs. throughput (produce)

Batching and compression (produce)

Consume throughput

Memory use (consume)

Watch for consume backpressure

Best Practices

Troubleshooting

Next Steps