Kafka component

The kafka component sends messages to and consumes messages from Apache Kafka topics, with consumer groups, keys, partitions, TLS and SASL.

Scheme: kafka: · Crate: cameleer-kafka-mqtt · Cargo feature: kafka-mqtt · Consumer: yes · Producer: yes · Apache Camel: Kafka

When to use it

• Consume events from a topic and process them (billing, enrichment, indexing) with a consumer group.
• Publish events from an API, a database poll or an MQTT bridge to a topic.
• Use Kafka as an AI agent's event bus: agents publish their actions and observations keyed by agent id, and other routes or agents react to them.
• Store and forward from the edge: buffer readings locally and push them to a central Kafka cluster.

The Kafka wire protocol client is built into Cameleer (no librdkafka, no JVM). It works with Apache Kafka, Redpanda and Kafka-compatible cloud services.

URI format

kafka:topic[?options]

Example

- route:
    id: orders-consumer
    from:
      uri: kafka:orders
      parameters:
        brokers: "{{env:KAFKA_BROKERS:localhost:9092}}"
        groupId: billing
        autoOffsetReset: earliest
      steps:
        - log: "partition ${header[kafka.PARTITION]} offset ${header[kafka.OFFSET]}: ${body}"
        - to: log:orders

The route joins the consumer group billing, starts at the oldest record when the group has no committed offset, and commits offsets automatically.

Agent event bus

- route:
    id: publish-agent-event
    from:
      uri: direct:agent-event
      steps:
        - setHeader:
            name: kafka.KEY
            simple: "${header.agentId}"
        - to: kafka:agent-events?brokers={{env:KAFKA_BROKERS:localhost:9092}}

Events of one agent share a key, so they land on the same partition and stay in order.

Rust DSL

use cameleer_core::prelude::*;

pub fn routes(ctx: &CamelContext) -> Result<()> {
    ctx.add_routes(|b| {
        b.from("kafka:orders?brokers=localhost:9092&groupId=billing&autoOffsetReset=earliest")
            .route_id("orders-consumer")
            .log("partition ${header[kafka.PARTITION]} offset ${header[kafka.OFFSET]}: ${body}")
            .to("log:orders");
        b.from("direct:agent-event")
            .route_id("publish-agent-event")
            .set_header("kafka.KEY", simple("${header.agentId}"))
            .to("kafka:agent-events?brokers=localhost:9092");
    })
}

Options

GSSAPI (Kerberos), OAUTHBEARER and client certificates (sslKeystoreLocation) are not supported. Other Camel Kafka tuning options are accepted and ignored.

Message headers

Other headers become Kafka record headers (strings as UTF-8, integers as 8-byte longs), except Camel*, org.apache.camel* and kafka.*. Received record headers become message headers (as bytes).

Run it with Docker

docker run --rm -e KAFKA_BROKERS=broker:9092 -v "$PWD:/routes:ro" pmsudhi/cameleer run /routes/orders.camel.yaml

Use --network host or the broker's network name so the container can reach Kafka.

Common questions

How do I read a topic from the beginning?

Set autoOffsetReset=earliest with a new groupId, or seekTo=beginning to ignore committed offsets.

How do I send one record per list element?

Send a list body: an array body sends one record per element.

How do I connect to Confluent Cloud or another SASL_SSL cluster?

securityProtocol=SASL_SSL&saslMechanism=PLAIN&saslJaasConfig=RAW(org.apache.kafka.common.security.plain.PlainLoginModule required username="{{env:KAFKA_KEY}}" password="{{env:KAFKA_SECRET}}";).

How do I commit offsets only after my processing succeeded?

Set autoCommitEnable=false&allowManualCommit=true and call the committer in CamelKafkaManualCommit from a processor, or keep auto-commit with breakOnFirstError=true.

Related

• paho (MQTT to Kafka bridge), seda, amqp
• Apache Camel Kafka component

Found a problem?

If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: open an issue that describes what you saw, then fork the repository, fix it, add or update a test, run cargo test --workspace, and open a pull request. CONTRIBUTING.md explains the ground rules.