Apache Camel: Kafka Component

The Camel Kafka component lets you produce to and consume from Kafka topics using the familiar Camel route DSL, integrating Kafka into any Camel integration route without writing Kafka producer/consumer code directly.

Apache Camel

Dependency
Consuming from Kafka
Producing to Kafka
Kafka Headers in Camel
Key Configuration Options
Spring Boot Configuration

Dependency

Add the camel-kafka-starter for Spring Boot projects, or camel-kafka for standalone Camel.

XML

<!-- Spring Boot -->
<dependency>
  <groupId>org.apache.camel.springboot</groupId>
  <artifactId>camel-kafka-starter</artifactId>
</dependency>

<!-- Standalone Camel (non-Spring Boot) -->
<dependency>
  <groupId>org.apache.camel</groupId>
  <artifactId>camel-kafka</artifactId>
</dependency>

Consuming from Kafka

Use from("kafka:<topic>?...") to start a route that consumes messages from a Kafka topic. The message body contains the Kafka record value; headers carry metadata.

Java

import org.apache.camel.builder.RouteBuilder;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumerRoute extends RouteBuilder {

    @Override
    public void configure() {
        from("kafka:orders"
                + "?brokers=localhost:9092"
                + "&groupId=camel-order-group"
                + "&autoOffsetReset=earliest"
                + "&maxPollRecords=50")
            .routeId("kafka-consumer")
            .log("Received from Kafka: ${body}")
            .to("direct:processOrder");

        from("direct:processOrder")
            .log("Processing order: ${body}");
    }
}

Camel automatically manages the Kafka consumer lifecycle: it starts the consumer when the route starts and shuts it down gracefully when the route stops. You do not need to manage consumer threads manually.

Producing to Kafka

Use to("kafka:<topic>?...") to publish the exchange body as a Kafka message. Set the message key via the KafkaConstants.KEY header before calling the Kafka endpoint.

Java

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.springframework.stereotype.Component;

@Component
public class KafkaProducerRoute extends RouteBuilder {

    @Override
    public void configure() {

        // Simple produce — body becomes the Kafka record value
        from("timer:trigger?period=5000")
            .setBody(constant("New order event"))
            .to("kafka:processed-orders?brokers=localhost:9092");

        // Produce with a message key
        from("direct:sendWithKey")
            .setHeader(KafkaConstants.KEY, header("orderId")) // use a route header as the Kafka key
            .to("kafka:processed-orders?brokers=localhost:9092");

        // Produce to a specific partition
        from("direct:sendToPartition")
            .setHeader(KafkaConstants.PARTITION_KEY, constant(0))
            .to("kafka:processed-orders?brokers=localhost:9092");
    }
}

Kafka Headers in Camel

When consuming from Kafka, Camel populates exchange headers with record metadata. When producing, you can set these headers to control how the record is written.

Header Constant	Direction	Description
`KafkaConstants.TOPIC`	Consume	The topic the message was received from.
`KafkaConstants.KEY`	Both	The Kafka record key (as a String by default).
`KafkaConstants.PARTITION_KEY`	Both	The partition number. Set when producing to target a specific partition.
`KafkaConstants.OFFSET`	Consume	The offset of the consumed record.
`KafkaConstants.TIMESTAMP`	Consume	The record timestamp in milliseconds since epoch.
`KafkaConstants.HEADERS`	Both	Kafka record headers (Map of key → byte[]).

Java

from("kafka:orders?brokers=localhost:9092&groupId=camel-group")
    .process(exchange -> {
        String topic     = exchange.getMessage().getHeader(KafkaConstants.TOPIC, String.class);
        String key       = exchange.getMessage().getHeader(KafkaConstants.KEY, String.class);
        Integer partition = exchange.getMessage().getHeader(KafkaConstants.PARTITION_KEY, Integer.class);
        Long offset      = exchange.getMessage().getHeader(KafkaConstants.OFFSET, Long.class);

        System.out.printf("topic=%s key=%s partition=%d offset=%d%n",
            topic, key, partition, offset);
    });

Key Configuration Options

Kafka component URI options map directly to Kafka consumer and producer config properties:

URI Option	Default	Description
`brokers`	-	Comma-separated list of bootstrap brokers (required).
`groupId`	-	Consumer group ID (required for consumers).
`autoOffsetReset`	`latest`	Where to start reading when no committed offset exists (`earliest` or `latest`).
`maxPollRecords`	500	Maximum records returned in a single poll call.
`seekTo`	-	Seek to `beginning` or `end` of topic on start.
`autoCommitEnable`	`true`	Enable/disable auto offset commit. Set to `false` for manual commit.
`concurrentConsumers`	1	Number of consumer threads (up to partition count).
`keyDeserializer`	StringDeserializer	Deserializer class for message keys.
`valueDeserializer`	StringDeserializer	Deserializer class for message values.

Spring Boot Configuration

Configure the Kafka component globally in application.properties using the camel.component.kafka.* prefix. Individual routes can still override these defaults via URI parameters.

Properties

# Component-level defaults — apply to all Kafka endpoints
camel.component.kafka.brokers=localhost:9092
camel.component.kafka.group-id=camel-app-group
camel.component.kafka.auto-offset-reset=earliest
camel.component.kafka.max-poll-records=100
camel.component.kafka.concurrent-consumers=3

# Security (optional)
camel.component.kafka.security-protocol=SASL_SSL
camel.component.kafka.sasl-mechanism=PLAIN
camel.component.kafka.sasl-jaas-config=org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="secret";

Java

// With component-level defaults set, the route URI becomes much shorter
@Component
public class OrderRoute extends RouteBuilder {
    @Override
    public void configure() {
        // brokers and groupId come from application.properties
        from("kafka:orders")
            .log("Received: ${body}")
            .to("kafka:processed-orders");
    }
}

Properties set in application.properties under camel.component.kafka.* become the component defaults. URI parameters on individual endpoints always take precedence over component defaults.

Contents

Dependency

Consuming from Kafka

Producing to Kafka

Kafka Headers in Camel

Key Configuration Options

Spring Boot Configuration

Related articles