How do you integrate Apache Camel with Apache Kafka?

Camel integrates with Kafka via the camel-kafka component, which wraps the native Kafka Java client. The URI scheme is kafka:topicName?brokers=...&options. It supports both consuming (from()) and producing (to()) messages, with full access to Kafka record metadata through Exchange headers.

    org.apache.camel
    camel-kafka


// Consumer: read from Kafka, process, produce to Kafka:
from("kafka:orders?brokers=localhost:9092&groupId=order-processor")
    .log("Received from partition ${header.kafka.PARTITION} offset ${header.kafka.OFFSET}")
    .unmarshal().json(Order.class)
    .process(new OrderProcessor())
    .marshal().json()
    .to("kafka:processed-orders?brokers=localhost:9092");

// application.properties for Camel Spring Boot:
camel.component.kafka.brokers=localhost:9092
camel.component.kafka.security-protocol=SASL_SSL
camel.component.kafka.sasl-mechanism=PLAIN

Key consumer options: groupId (consumer group), autoOffsetReset (earliest/latest), maxPollRecords, pollTimeoutMs. Key producer options: partitionKey, compressionCodec. The component automatically sets headers: kafka.TOPIC, kafka.PARTITION, kafka.OFFSET, kafka.KEY. Use seekTo=beginning for replay. Manual offset commit is supported via allow.auto.create.topics and the KafkaManualCommit header.