Since we're on a major migration process of this website, some component documents here are out of sync right now. In the meantime you may want to look at the early version of the new website
https://camel.apache.org/staging/
We would very much like to receive any feedback on the new site, please join the discussion on the Camel user mailing list.
Kafka ComponentAvailable as of Camel 2.13 The kafka: component is used for communicating with Apache Kafka message broker. Maven users will need to add the following dependency to their <dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-kafka</artifactId>
<version>x.x.x</version>
<!-- use the same version as your Camel core version -->
</dependency>
Camel 2.17 or newerScala is no longer used, as we use the kafka java client. Camel 2.16 or olderAnd then the Scala libraries of choice. camel-kafka does not include that dependency, but assumes it is provided. For example to use Scala 2.10.4 add: <dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.10.4</version>
</dependency>
URI formatkafka:server:port[?options]
Options (Camel 2.16 or older)
You can append query options to the URI in the following format, Producer Options (Camel 2.16 or older)
Consumer Options (Camel 2.16 or older)
Options (Camel 2.17 or newer)
Producer Options (Camel 2.17 or newer)
Consumer Options (Camel 2.17 or newer)
SamplesCamel 2.16 or olderConsuming messages: from("kafka:localhost:9092?topic=test&zookeeperHost=localhost&zookeeperPort=2181&groupId=group1").to("log:input");
Producing messages: See unit tests of camel-kafka for more examples Camel 2.17 or newerConsuming messages: from("kafka:localhost:9092?topic=test&groupId=testing&autoOffsetReset=earliest&consumersCount=1")
.process(new Processor() {
@Override
public void process(Exchange exchange)
throws Exception {
String messageKey = "";
if (exchange.getIn() != null) {
Message message = exchange.getIn();
Integer partitionId = (Integer) message
.getHeader(KafkaConstants.PARTITION);
String topicName = (String) message
.getHeader(KafkaConstants.TOPIC);
if (message.getHeader(KafkaConstants.KEY) != null)
messageKey = (String) message
.getHeader(KafkaConstants.KEY);
Object data = message.getBody();
System.out.println("topicName :: "
+ topicName + " partitionId :: "
+ partitionId + " messageKey :: "
+ messageKey + " message :: "
+ data + "\n");
}
}
}).to("log:input");
Producing messages: from("direct:start").process(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
exchange.getIn().setBody("Test Message from Camel Kafka Component Final",String.class);
exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, 0);
exchange.getIn().setHeader(KafkaConstants.KEY, "1");
}
}).to("kafka:localhost:9092?topic=test");
Using the Kafka idempotent repository (Available from Camel 2.19)The The topic used must be unique per idempotent repository instance. The mechanism does not have any requirements about the number of topic partitions; as the repository consumes from all partitions at the same time. It also does not have any requirements about the replication factor of the topic. Each repository instance that uses the topic (e.g. typically on different machines running in parallel) controls its own consumer group, so in a cluster of 10 Camel processes using the same topic each will control its own offset. On startup, the instance subscribes to the topic and rewinds the offset to the beginning, rebuilding the cache to the latest state. The cache will not be considered warmed up until one poll of A
The repository can be instantiated by defining the topic and To use, this repository must be placed in the Camel registry, either manually or by registration as a bean in Spring/Blueprint, as it is Sample usage is as follows: KafkaIdempotentRepository kafkaIdempotentRepository = new KafkaIdempotentRepository("idempotent-db-inserts", "localhost:9091");
SimpleRegistry registry = new SimpleRegistry();
registry.put("insertDbIdemRepo", kafkaIdempotentRepository); // must be registered in the registry, to enable access to the CamelContext
CamelContext context = new CamelContext(registry);
// later in RouteBuilder...
from("direct:performInsert")
.idempotentConsumer(header("id")).messageIdRepositoryRef("insertDbIdemRepo")
// once-only insert into database
.end()
In XML: <!-- simple -->
<bean id="insertDbIdemRepo" class="org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository">
<property name="topic" value="idempotent-db-inserts"/>
<property name="bootstrapServers" value="localhost:9091"/>
</bean>
<!-- complex -->
<bean id="insertDbIdemRepo" class="org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository">
<property name="topic" value="idempotent-db-inserts"/>
<property name="maxCacheSize" value="10000"/>
<property name="consumerConfig">
<props>
<prop key="bootstrap.servers">localhost:9091</prop>
</props>
</property>
<property name="producerConfig">
<props>
<prop key="bootstrap.servers">localhost:9091</prop>
</props>
</property>
</bean>
EndpointsCamel supports the Message Endpoint pattern using the Endpoint interface. Endpoints are usually created by a Component and Endpoints are usually referred to in the DSL via their URIs. From an Endpoint you can use the following methods
See Also |
