Red Hat SummitFuse 6 is no longer supported
As of February 2025, Red Hat Fuse 6 is no longer supported. If you are using Fuse 6, please upgrade to Red Hat build of Apache Camel.Este contenido no está disponible en el idioma seleccionado.
Chapter 90. Kafka
Kafka Component
Copiar enlaceEnlace copiado en el portapapeles!
Available 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
pom.xml for this component:
Camel 2.17 or newer: Scala is no longer used, as we use the Kafka Java client.
Camel 2.16 or older: You must also add a Maven dependency for your chosen Scala library. camel-kafka does not include that dependency, but assumes its 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>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.10.4</version>
</dependency>URI format
Copiar enlaceEnlace copiado en el portapapeles!
kafka:server:port[?options]
kafka:server:port[?options]Options
Copiar enlaceEnlace copiado en el portapapeles!
|
Property
|
Default
|
Description
|
|---|---|---|
zookeeperHost
|
|
The zookeeper host to use
|
zookeeperPort
|
2181
|
The zookeeper port to use
|
zookeeperConnect
|
Camel 2.13.3/2.14.1: If in use, then zookeeperHost/zookeeperPort is not used. | |
topic
|
|
The topic to use
|
groupId
|
||
partitioner
|
||
consumerStreams
|
10 | |
clientId
|
||
zookeeperSessionTimeoutMs
|
||
zookeeperConnectionTimeoutMs
|
||
zookeeperSyncTimeMs
|
||
consumersCount
|
1
|
Camel 2.15.0: The number of consumers that connect to Kafka server. |
batchSize
|
100
|
Camel 2.15.0: The batchSize that the BatchingConsumerTask processes once.
|
barrierAwaitTimeoutMs
|
10000
|
Camel 2.15.0: If the BatchingConsumerTask processes exchange exceed the batchSize, it will wait for barrierAwaitTimeoutMs.
|
bridgeEndpoint
|
false
|
Camel 2.16.0: If the bridgeEndpoint is true, the producer will ignore the topic header setting of the message.
|
You can append query options to the URI in the following format,
?option=value&option=value&...
Producer Options
Copiar enlaceEnlace copiado en el portapapeles!
|
Property
|
Default
|
Description
|
|---|---|---|
producerType
|
sync
|
Can have the following values:
|
compressionCodec
|
||
compressedTopics
|
||
messageSendMaxRetries
|
||
retryBackoffMs
|
||
topicMetadataRefreshIntervalMs
|
||
sendBufferBytes
|
||
requestRequiredAcks
|
||
requestTimeoutMs
|
||
queueBufferingMaxMs
|
||
queueBufferingMaxMessages
|
||
queueEnqueueTimeoutMs
|
||
batchNumMessages
|
||
serializerClass
|
||
keySerializerClass
|
Consumer Options
Copiar enlaceEnlace copiado en el portapapeles!
|
Property
|
Default
|
Description
|
|---|---|---|
consumerId
|
|
|
socketTimeoutMs
|
||
socketReceiveBufferBytes
|
||
fetchMessageMaxBytes
|
||
autoCommitEnable
|
||
autoCommitIntervalMs
|
||
queuedMaxMessages
|
||
rebalanceMaxRetries
|
||
fetchMinBytes
|
||
fetchWaitMaxMs
|
||
rebalanceBackoffMs
|
||
refreshLeaderBackoffMs
|
||
autoOffsetReset
|
||
consumerTimeoutMs
|
Samples
Copiar enlaceEnlace copiado en el portapapeles!
Consuming messages:
from("kafka:localhost:9092?topic=test&zookeeperHost=localhost&zookeeperPort=2181&groupId=group1").to("log:input");
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
Endpoints
Copiar enlaceEnlace copiado en el portapapeles!
Camel 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
- createProducer() will create a Producer for sending message exchanges to the endpoint
- createConsumer() implements the Event Driven Consumer pattern for consuming message exchanges from the endpoint via a Processor when creating a Consumer
- createPollingConsumer() implements the Polling Consumer pattern for consuming message exchanges from the endpoint via a PollingConsumer
See Also
Copiar enlaceEnlace copiado en el portapapeles!
'%20d='M201.5%20305.5c-13.8%200-24.9-11.1-24.9-24.6%200-13.8%2011.1-24.9%2024.9-24.9%2013.6%200%2024.6%2011.1%2024.6%2024.9%200%2013.6-11.1%2024.6-24.6%2024.6zM504%20256c0%20137-111%20248-248%20248S8%20393%208%20256%20119%208%20256%208s248%20111%20248%20248zm-132.3-41.2c-9.4%200-17.7%203.9-23.8%2010-22.4-15.5-52.6-25.5-86.1-26.6l17.4-78.3%2055.4%2012.5c0%2013.6%2011.1%2024.6%2024.6%2024.6%2013.8%200%2024.9-11.3%2024.9-24.9s-11.1-24.9-24.9-24.9c-9.7%200-18%205.8-22.1%2013.8l-61.2-13.6c-3-.8-6.1%201.4-6.9%204.4l-19.1%2086.4c-33.2%201.4-63.1%2011.3-85.5%2026.8-6.1-6.4-14.7-10.2-24.1-10.2-34.9%200-46.3%2046.9-14.4%2062.8-1.1%205-1.7%2010.2-1.7%2015.5%200%2052.6%2059.2%2095.2%20132%2095.2%2073.1%200%20132.3-42.6%20132.3-95.2%200-5.3-.6-10.8-1.9-15.8%2031.3-16%2019.8-62.5-14.9-62.5zM302.8%20331c-18.2%2018.2-76.1%2017.9-93.6%200-2.2-2.2-6.1-2.2-8.3%200-2.5%202.5-2.5%206.4%200%208.6%2022.8%2022.8%2087.3%2022.8%20110.2%200%202.5-2.2%202.5-6.1%200-8.6-2.2-2.2-6.1-2.2-8.3%200zm7.7-75c-13.6%200-24.6%2011.1-24.6%2024.9%200%2013.6%2011.1%2024.6%2024.6%2024.6%2013.8%200%2024.9-11.1%2024.9-24.6%200-13.8-11-24.9-24.9-24.9z'/%3e%3c/svg%3e){kind=small}