Modified: websites/production/camel/content/idempotent-consumer.html
==============================================================================
--- websites/production/camel/content/idempotent-consumer.html (original)
+++ websites/production/camel/content/idempotent-consumer.html Tue Mar 7
15:20:59 2017
@@ -86,7 +86,7 @@
<tbody>
<tr>
<td valign="top" width="100%">
-<div class="wiki-content maincontent"><h3
id="IdempotentConsumer-IdempotentConsumer">Idempotent Consumer</h3><p>The <a
shape="rect" class="external-link"
href="http://www.enterpriseintegrationpatterns.com/IdempotentReceiver.html";
rel="nofollow">Idempotent Consumer</a> from the <a shape="rect"
href="enterprise-integration-patterns.html">EIP patterns</a> is used to filter
out duplicate messages.</p><p>This pattern is implemented using the <a
shape="rect" class="external-link"
href="http://camel.apache.org/maven/current/camel-core/apidocs/org/apache/camel/processor/idempotent/IdempotentConsumer.html";>IdempotentConsumer</a>
class. This uses an <a shape="rect" href="expression.html">Expression</a> to
calculate a unique message ID string for a given message exchange; this ID can
then be looked up in the <a shape="rect" class="external-link"
href="http://camel.apache.org/maven/current/camel-core/apidocs/org/apache/camel/spi/IdempotentRepository.html";>IdempotentRepository</a>
to see if it h
as been seen before; if it has the message is consumed; if its not then the
message is processed and the ID is added to the repository.</p><p>The
Idempotent Consumer essentially acts like a <a shape="rect"
href="message-filter.html">Message Filter</a> to filter out
duplicates.</p><p>Camel will add the message id eagerly to the repository to
detect duplication also for Exchanges currently in progress.<br clear="none">
On completion Camel will remove the message id from the repository if the
Exchange failed, otherwise it stays there.</p><p>Camel provides the following
Idempotent Consumer implementations:</p><ul
class="alternate"><li>MemoryIdempotentRepository</li><li><a shape="rect"
href="file2.html">FileIdempotentRepository</a></li><li><a shape="rect"
href="hazelcast-component.html">HazelcastIdempotentRepository</a>
(<strong>Available as of Camel 2.8</strong>)</li><li><a shape="rect"
href="sql-component.html">JdbcMessageIdRepository</a> (<strong>Available as of
Camel 2.7</strong>)</l
i><li><a shape="rect" href="jpa.html">JpaMessageIdRepository</a></li><li><p><a
shape="rect" href="infinispan.html">InfinispanIdempotentRepository</a>
(<strong>Available as of Camel 2.13.0)</strong></p></li><li><p><a shape="rect"
href="jcache.html">JCacheIdempotentRepository</a><strong> (<strong>Available
as of Camel 2.17.0)</strong></strong></p></li><li><p><a shape="rect"
href="spring.html">SpringCacheIdempotentRepository</a> <strong>(<strong>Available
as of Camel 2.17.1)</strong></strong><strong><strong><br
clear="none"></strong></strong></p></li><li><p><strong><strong><a shape="rect"
href="ehcache.html">EhcacheIdempotentRepository</a> <strong>(<strong>Available
as of Camel 2.18.0)</strong></strong><br
clear="none"></strong></strong></p></li></ul><h3
id="IdempotentConsumer-Options">Options</h3><p>The Idempotent Consumer has the
following options:</p><div class="table-wrap"><table
class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh"><p>O
ption</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Default</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Description</p></th></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>eager</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>true</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>Eager controls whether Camel adds the message to the
repository before or after the exchange has been processed. If enabled before
then Camel will be able to detect duplicate messages even when messages are
currently in progress. By disabling Camel will only detect duplicates when a
message has successfully been processed.</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>messageIdRepositoryRef</p></td><td
colspan="1" rowspan="1" class="confluenceTd"><p><code>null</code></p></td><td
colspan="1" rowspan="1" class="confluenceTd"><p>A reference to a
<code>IdempotentRepository</code> to lookup in the registry. This option is man
datory when using XML DSL.</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>skipDuplicate</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>true</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.8:</strong> Sets whether to skip
duplicate messages. If set to <code>false</code> then the message will be
continued. However the <a shape="rect" href="exchange.html">Exchange</a> has
been marked as a duplicate by having the <code>Exchange.DUPLICATE_MESSAG</code>
exchange property set to a <code>Boolean.TRUE</code>
value.</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>removeOnFailure</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>true</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.9:</strong> Sets whether to remove the
id of an Exchange that failed.</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">completionEager</td><td colspan="1" rowspan="1"
class="confl
uenceTd">false</td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.16:</strong> Sets whether to complete
the idempotent consumer eager or when the exchange is done.</p><p>If this
option is true to complete eager, then the idempotent consumer will trigger its
completion when the exchange reached the end of the block of the idempotent
consumer pattern. So if the exchange is continued routed after the block ends,
then whatever happens there does not affect the state.</p><p>If this option is
false (default) to not complete eager, then the idempotent consumer will
complete when the exchange is done being routed. So if the exchange is
continued routed after the block ends, then whatever happens there also affect
the state. For example if the exchange failed due to an exception, then the
state of the idempotent consumer will be a
rollback.</p></td></tr></tbody></table></div><h3
id="IdempotentConsumer-Usingthe"><strong>Using the <a shape="rect"
href="fluent-builders.html
">Fluent Builders</a></strong></h3><p>The following example will use the
header <strong>myMessageId</strong> to filter out duplicates</p><div
class="code panel pdl" style="border-width: 1px;"><div class="codeContent
panelContent pdl">
+<div class="wiki-content maincontent"><h3
id="IdempotentConsumer-IdempotentConsumer">Idempotent Consumer</h3><p>The <a
shape="rect" class="external-link"
href="http://www.enterpriseintegrationpatterns.com/IdempotentReceiver.html";
rel="nofollow">Idempotent Consumer</a> from the <a shape="rect"
href="enterprise-integration-patterns.html">EIP patterns</a> is used to filter
out duplicate messages.</p><p>This pattern is implemented using the <a
shape="rect" class="external-link"
href="http://camel.apache.org/maven/current/camel-core/apidocs/org/apache/camel/processor/idempotent/IdempotentConsumer.html";>IdempotentConsumer</a>
class. This uses an <a shape="rect" href="expression.html">Expression</a> to
calculate a unique message ID string for a given message exchange; this ID can
then be looked up in the <a shape="rect" class="external-link"
href="http://camel.apache.org/maven/current/camel-core/apidocs/org/apache/camel/spi/IdempotentRepository.html";>IdempotentRepository</a>
to see if it h
as been seen before; if it has the message is consumed; if its not then the
message is processed and the ID is added to the repository.</p><p>The
Idempotent Consumer essentially acts like a <a shape="rect"
href="message-filter.html">Message Filter</a> to filter out
duplicates.</p><p>Camel will add the message id eagerly to the repository to
detect duplication also for Exchanges currently in progress.<br clear="none">
On completion Camel will remove the message id from the repository if the
Exchange failed, otherwise it stays there.</p><p>Camel provides the following
Idempotent Consumer implementations:</p><ul
class="alternate"><li>MemoryIdempotentRepository</li><li><a shape="rect"
href="file2.html">FileIdempotentRepository</a></li><li><a shape="rect"
href="hazelcast-component.html">HazelcastIdempotentRepository</a>
(<strong>Available as of Camel 2.8</strong>)</li><li><a shape="rect"
href="sql-component.html">JdbcMessageIdRepository</a> (<strong>Available as of
Camel 2.7</strong>)</l
i><li><a shape="rect" href="jpa.html">JpaMessageIdRepository</a></li><li><p><a
shape="rect" href="infinispan.html">InfinispanIdempotentRepository</a>
(<strong>Available as of Camel 2.13.0)</strong></p></li><li><p><a shape="rect"
href="jcache.html">JCacheIdempotentRepository</a><strong> (<strong>Available
as of Camel 2.17.0)</strong></strong></p></li><li><p><a shape="rect"
href="spring.html">SpringCacheIdempotentRepository</a> <strong>(<strong>Available
as of Camel 2.17.1)</strong></strong><strong><strong><br
clear="none"></strong></strong></p></li><li><p><a shape="rect"
href="ehcache.html">EhcacheIdempotentRepository</a><strong><strong> <strong>(<strong>Available
as of Camel 2.18.0)</strong></strong></strong></strong></p></li><li><a
shape="rect" href="kafka.html">KafkaIdempotentRepository</a> (<strong>Available
as of Camel 2.19.0)</strong></li></ul><h3
id="IdempotentConsumer-Options">Options</h3><p>The Idempotent Consumer has the
following options:</p><div class="tabl
e-wrap"><table class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh"><p>Option</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Default</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Description</p></th></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>eager</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>true</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>Eager controls whether Camel adds the message to the
repository before or after the exchange has been processed. If enabled before
then Camel will be able to detect duplicate messages even when messages are
currently in progress. By disabling Camel will only detect duplicates when a
message has successfully been processed.</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>messageIdRepositoryRef</p></td><td
colspan="1" rowspan="1" class="confluenceTd"><p><code>null</code></p></td><td
colspan="1" rowspan="1" class="confluence
Td"><p>A reference to a <code>IdempotentRepository</code> to lookup in the
registry. This option is mandatory when using XML DSL.</p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>skipDuplicate</p></td><td
colspan="1" rowspan="1" class="confluenceTd"><p>true</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p><strong>Camel 2.8:</strong> Sets whether to
skip duplicate messages. If set to <code>false</code> then the message will be
continued. However the <a shape="rect" href="exchange.html">Exchange</a> has
been marked as a duplicate by having the <code>Exchange.DUPLICATE_MESSAG</code>
exchange property set to a <code>Boolean.TRUE</code>
value.</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>removeOnFailure</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>true</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.9:</strong> Sets whether to remove the
id of an Exchange that failed.</p></td></tr><tr><td co
lspan="1" rowspan="1" class="confluenceTd">completionEager</td><td colspan="1"
rowspan="1" class="confluenceTd">false</td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.16:</strong> Sets whether to complete
the idempotent consumer eager or when the exchange is done.</p><p>If this
option is true to complete eager, then the idempotent consumer will trigger its
completion when the exchange reached the end of the block of the idempotent
consumer pattern. So if the exchange is continued routed after the block ends,
then whatever happens there does not affect the state.</p><p>If this option is
false (default) to not complete eager, then the idempotent consumer will
complete when the exchange is done being routed. So if the exchange is
continued routed after the block ends, then whatever happens there also affect
the state. For example if the exchange failed due to an exception, then the
state of the idempotent consumer will be a
rollback.</p></td></tr></tbody></table>
</div><h3 id="IdempotentConsumer-Usingthe"><strong>Using the <a shape="rect"
href="fluent-builders.html">Fluent Builders</a></strong></h3><p>The following
example will use the header <strong>myMessageId</strong> to filter out
duplicates</p><div class="code panel pdl" style="border-width: 1px;"><div
class="codeContent panelContent pdl">
<script class="brush: java; gutter: false; theme: Default"
type="syntaxhighlighter"><![CDATA[
RouteBuilder builder = new RouteBuilder() {
public void configure() {
Modified: websites/production/camel/content/kafka.html
==============================================================================
--- websites/production/camel/content/kafka.html (original)
+++ websites/production/camel/content/kafka.html Tue Mar 7 15:20:59 2017
@@ -102,13 +102,16 @@
</div></div><h3 id="Kafka-URIformat">URI format</h3><div class="code panel
pdl" style="border-width: 1px;"><div class="codeContent panelContent pdl">
<script class="brush: java; gutter: false; theme: Default"
type="syntaxhighlighter"><![CDATA[kafka:server:port[?options]
]]></script>
-</div></div><p> </p><h3 id="Kafka-Options(Camel2.16orolder)">Options
(Camel 2.16 or older)</h3><div class="confluenceTableSmall"><div
class="table-wrap">
- <table class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh"><p>Property</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Default</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Description</p></th></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>zookeeperHost</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p> </p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>The zookeeper host to use</p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>zookeeperPort</p></td><td
colspan="1" rowspan="1" class="confluenceTd"><p>2181</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p>The zookeeper port to
use</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">zookeeperConnect</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"><strong>Camel 2.13.3/2.14.1:</strong> If in use, then
zookeeperHost/zookeeperPort is no
t used.</td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>topic</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p> </p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>The topic to use</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>groupId</p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>partitioner</p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>consumerStreams</p></td><td colspan="1" rowspan="1"
class="confluenceTd">10</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>clientId</p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1" c
lass="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>zookeeperSessionTimeoutMs</p></td><td colspan="1"
rowspan="1" class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>zookeeperConnectionTimeoutMs</p></td><td colspan="1"
rowspan="1" class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>zookeeperSyncTimeMs</p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p><span style="color:
rgb(51,51,51);">consumersCount</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><span style="color:
rgb(51,51,51);">1</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.15.0:</strong> The numbe
r of consumers that connect to kafka server</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p><span style="color:
rgb(51,51,51);">batchSize</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><span style="color:
rgb(51,51,51);">100</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.15.0: </strong>The batchSize that the
BatchingConsumerTask processes once, deprecated since <strong>2.17.1</strong>,
removed<strong><br clear="none"></strong>since
<strong>2.18.0</strong></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p><span style="color:
rgb(51,51,51);">barrierAwaitTimeoutMs</span></p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p><span style="color:
rgb(51,51,51);">10000</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.15.0: </strong>If the
BatchingConsumerTask processes exchange exceed the batchSize, it will wait for
<span style="color: rgb(51,51,51);">barr
ierAwaitTimeoutMs</span>, deprecated since <strong>2.17.1</strong>, removed
since <strong>2.18.0</strong><strong>.<br
clear="none"></strong></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">bridgeEndpoint</td><td colspan="1" rowspan="1"
class="confluenceTd">false</td><td colspan="1" rowspan="1"
class="confluenceTd">Camel 2.16.0: If bridgeEndpoint is true, the producer will
ignore the topic header setting of the message.</td></tr></tbody></table>
-</div></div><p>You can append query options to the URI in the following
format, <code>?option=value&option=value&...</code></p><h3
id="Kafka-ProducerOptions(Camel2.16orolder)">Producer Options (Camel 2.16
or older)</h3><div class="confluenceTableSmall"><div class="table-wrap">
- <table class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh"><p>Property</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Default</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Description</p></th></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>producerType</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p>sync (Taken from native KafkaProducer
class)</p></td><td colspan="1" rowspan="1" class="confluenceTd"><p>sync - send
message/batch immediately, and wait until response is received</p><p>async -
queue the message/batch to send. There is a thread per broker (Kafka node)
which polls from this queue upon <span>queueBufferingMaxMs or
<span>batchNumMessages</span></span></p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd">compressionCodec</td><td colspan="1"
rowspan="1" class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">compressedTopics</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">messageSendMaxRetries</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">retryBackoffMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">topicMetadataRefreshIntervalMs</td><td colspan="1"
rowspan="1" class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">sendBufferBytes</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="
1" rowspan="1" class="confluenceTd">requestRequiredAcks</td><td colspan="1"
rowspan="1" class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">requestTimeoutMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">queueBufferingMaxMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">queueBufferingMaxMessages</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">queueEnqueueTimeoutMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr
><tr><td colspan="1" rowspan="1" class="confluenceTd">batchNumMessages</td><td
>colspan="1" rowspan="1" class="confluenceTd"> </td><td colspan="1"
>rowspan="1" class="confluenceTd"> </td></tr><tr><td colspan="1"
>rowspan="1" class="confluenceTd">serializerClass</td><td colspan="1"
>rowspan="1" class="confluenceTd"> </td><td colspan="1" rowspan="1"
>class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
>class="confluenceTd">keySerializerClass</td><td colspan="1" rowspan="1"
>class="confluenceTd"> </td><td colspan="1" rowspan="1"
>class="confluenceTd"> </td></tr></tbody></table>
-</div></div><h3 id="Kafka-ConsumerOptions(Camel2.16orolder)">Consumer
Options (Camel 2.16 or older)</h3><div class="confluenceTableSmall"><div
class="table-wrap">
- <table class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh"><p>Property</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Default</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Description</p></th></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>consumerId</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p> </p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">socketTimeoutMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">socketReceiveBufferBytes</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">fetchMessageMaxBytes</td><td colspan="1" rowspan="1"
class="confluenceTd">&
#160;</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">autoCommitEnable</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">autoCommitIntervalMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">queuedMaxMessages</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">rebalanceMaxRetries</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">fetchMinBytes</td><td colspan="1" rowspan="1"
class="confluenceTd
"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">fetchWaitMaxMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">rebalanceBackoffMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">refreshLeaderBackoffMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">autoOffsetReset</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">consumerTimeoutMs</td><td colspan="1" rowspan="1"
class="confluen
ceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr></tbody></table>
-</div></div><h3 id="Kafka-Options(Camel2.17ornewer)">Options (Camel 2.17 or
newer)</h3><div class="table-wrap"><table
class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh">Property</th><th colspan="1" rowspan="1"
class="confluenceTh">Default</th><th colspan="1" rowspan="1"
class="confluenceTh">Description</th></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span class="s1">topic</span></p></td><td
colspan="1" rowspan="1" class="confluenceTd"> </td><td colspan="1"
rowspan="1" class="confluenceTd">Topic to use. From the
<strong>consumer</strong> side you can specify also a comma separated list of
topics.</td></tr><tr><td colspan="1" rowspan="1" class="confluenceTd"><p
class="p1"><span class="s1">groupId</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span class="s1">con
sumerStreams</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd">10</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span class="s1">clientId</span></p></td><td
colspan="1" rowspan="1" class="confluenceTd"> </td><td colspan="1"
rowspan="1" class="confluenceTd"> </td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p class="p1"><span
class="s1">consumersCount</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd">1</td><td colspan="1" rowspan="1" class="confluenceTd"><p
class="p1"><span class="s1">The number of consumers that connect to kafka
server</span></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span
class="s1">batchSize</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd">100</td><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span class="s1">Commit Size if auto commit
is false</span></p><
/td></tr><tr><td colspan="1" rowspan="1" class="confluenceTd"><p
class="p1"><span class="s1">bridgeEndpoint</span></p></td><td colspan="1"
rowspan="1" class="confluenceTd">false</td><td colspan="1" rowspan="1"
class="confluenceTd"><span>If the bridgeEndpoint is true, the producer will
ignore the topic header setting of the
message.</span></td></tr></tbody></table></div><p> </p><h3
id="Kafka-ProducerOptions(Camel2.17ornewer)">Producer Options (Camel 2.17
or newer)</h3><div class="table-wrap"><table
class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh">Property</th><th colspan="1" rowspan="1"
class="confluenceTh">Default & Description Reference</th></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>serializerClass</p></td><td
colspan="1" rowspan="46" class="confluenceTd"><p class="p1"><span
class="s1"><span class="nolink"><a shape="rect" class="external-link"
href="http://kafka.apache.org/documentation.html#producerconfigs";>http://
kafka.apache.org/documentation.html#producerconfigs</a></span></span></p><p
class="p1"><span class="s1"><span class="nolink">serializerClass : <span
class="pl-s">org.apache.kafka.common.serialization.StringSerializer</span></span></span></p><p
class="p1"><span class="s1"><span class="nolink">keySerializerClass : <span
class="s1"><span class="nolink"><span
class="pl-s">org.apache.kafka.common.serialization.StringSerializer</span></span></span></span></span></p><p>partitioner
: <span
class="pl-s">org.apache.kafka.clients.producer.internals.DefaultPartitioner<span
class="pl-pds"> </span></span></p><p> </p><p> </p><p
class="p1"><span class="s1"><span class="nolink"><br
clear="none"></span></span></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>keySerializerClass</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>requestRequiredAcks </p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>bufferMemorySize </p></
td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>compressionCodec </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>retries </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeyPassword</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslKeystoreLocation</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeystorePassword</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslTruststoreLocation</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslTruststorePassword</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>producerBatchSize </p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>clientId</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>connectionMaxIdleMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>lingerMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>maxBlockMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>maxRequestSize </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>partitioner </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>receiveBufferBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>requestTimeoutMs </p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>saslKerberosServiceName</p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>saslMechanism <strong>(from
Camel 2.18)</strong></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>securityProtocol</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sendBufferBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslEnabledProtocols </p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>sslKeystoreType </p></td
></tr><tr><td colspan="1" rowspan="1"
>class="confluenceTd"><p>sslProtocol </p></td></tr><tr><td colspan="1"
>rowspan="1" class="confluenceTd"><p>sslProvider</p></td></tr><tr><td
>colspan="1" rowspan="1"
>class="confluenceTd"><p>sslTruststoreType</p></td></tr><tr><td colspan="1"
>rowspan="1"
>class="confluenceTd"><p>maxInFlightRequest </p></td></tr><tr><td
>colspan="1" rowspan="1"
>class="confluenceTd"><p>metadataMaxAgeMs </p></td></tr><tr><td
>colspan="1" rowspan="1"
>class="confluenceTd"><p>metricReporters</p></td></tr><tr><td colspan="1"
>rowspan="1"
>class="confluenceTd"><p>noOfMetricsSample </p></td></tr><tr><td
>colspan="1" rowspan="1"
>class="confluenceTd"><p>metricsSampleWindowMs </p></td></tr><tr><td
>colspan="1" rowspan="1"
>class="confluenceTd"><p>reconnectBackoffMs </p></td></tr><tr><td
>colspan="1" rowspan="1"
>class="confluenceTd"><p>retryBackoffMs </p></td></tr><tr><td colspan="1"
>rowspan="1" class="confluenceTd"><p>kerberosInitCmd </p></td></tr>
<tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosBeforeReloginMinTime </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosRenewJitter </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosRenewWindowFactor </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslCipherSuites</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslEndpointAlgorithm</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeymanagerAlgorithm </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslTrustmanagerAlgorithm </p></td></tr></tbody></table></div><h3
id="Kafka-ConsumerOptions(Camel2.17ornewer)">Consumer Options (Camel 2.17
or newer)</h3><div class="table-wrap"><table
class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh">Property</th><th colspan="1" rowspan="1"
class="confluenceTh">Default & Description Re
ference</th></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p> </p></td><td colspan="1" rowspan="45"
class="confluenceTd"><p class="p1"><span class="s1"><span class="nolink"><a
shape="rect" class="external-link"
href="http://kafka.apache.org/documentation.html#newconsumerconfigs";>http://kafka.apache.org/documentation.html#newconsumerconfigs</a></span></span></p><p
class="p1"><span class="s1"><span class="nolink">keyDeserializer : <span
class="pl-s">org.apache.kafka.common.serialization.StringDeserializer</span></span></span></p><p
class="p1"><span class="s1"><span class="nolink">valueDeserializer : <span
class="s1"><span class="nolink"><span
class="pl-s">org.apache.kafka.common.serialization.StringDeserializer</span></span></span></span></span></p><p>partitionAssignor
: <span
class="pl-s">org.apache.kafka.clients.consumer.RangeAssignor</span></p><p> </p><p
class="p1"><span class="s1"><span class="nolink"><span class="s1"><span
class="nolink"><span class="pl-s"><
br clear="none"></span></span></span></span></span></p><p class="p1"><span
class="s1"><span class="nolink"><br
clear="none"></span></span></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>keyDeserializer</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>valueDeserializer</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>fetchMinBytes </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>groupId</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>heartbeatIntervalMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>maxPartitionFetchBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sessionTimeoutMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslKeyPassword</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeystoreLocation</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluence
Td"><p>sslKeystorePassword</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>sslTruststoreLocation</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslTruststorePassword</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>autoOffsetReset </p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>connectionMaxIdleMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>autoCommitEnable </p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>partitionAssignor </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>receiveBufferBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>consumerRequestTimeoutMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>saslKerberosServiceName</p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p><span>saslMechanism
</span><strong>(from Camel 2.18)</strong>
</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>securityProtocol</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sendBufferBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslEnabledProtocols </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeystoreType </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslProtocol </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslProvider</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslTruststoreType</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>autoCommitIntervalMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>checkCrcs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>clientId</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>fetchWaitMaxMs </p></td></tr><tr><td
colspan=
"1" rowspan="1"
class="confluenceTd"><p>metadataMaxAgeMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>metricReporters</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>noOfMetricsSample </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>metricsSampleWindowMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>reconnectBackoffMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>retryBackoffMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>kerberosInitCmd </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosBeforeReloginMinTime </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosRenewJitter </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosRenewWindowFactor </p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>sslCipherSuites
</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>sslEndpointAlgorithm</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>sslKeymanagerAlgorithm </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslTrustmanagerAlgorithm </p></td></tr></tbody></table></div><p> </p><h3
id="Kafka-Samples">Samples</h3><h4 id="Kafka-Camel2.16orolder.1">Camel 2.16 or
older</h4><p>Consuming messages:</p><div class="code panel pdl"
style="border-width: 1px;"><div class="codeContent panelContent pdl">
+</div></div><p> </p><h3 id="Kafka-Options(Camel2.16orolder)">Options
(Camel 2.16 or older)</h3><div class="confluenceTableSmall"><div
class="table-wrap"><table class="confluenceTable"><tbody><tr><th colspan="1"
rowspan="1" class="confluenceTh"><p>Property</p></th><th colspan="1"
rowspan="1" class="confluenceTh"><p>Default</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Description</p></th></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>zookeeperHost</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p> </p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>The zookeeper host to use</p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>zookeeperPort</p></td><td
colspan="1" rowspan="1" class="confluenceTd"><p>2181</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p>The zookeeper port to
use</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">zookeeperConnect</td><td colspan="1" rowspan="1"
class="conf
luenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"><strong>Camel 2.13.3/2.14.1:</strong> If in use, then
zookeeperHost/zookeeperPort is not used.</td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>topic</p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p> </p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>The topic to use</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>groupId</p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>partitioner</p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>consumerStreams</p></td><td colspan="1" rowspan="1"
class="confluenceTd">10</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><
td colspan="1" rowspan="1" class="confluenceTd"><p>clientId</p></td><td
colspan="1" rowspan="1" class="confluenceTd"> </td><td colspan="1"
rowspan="1" class="confluenceTd"> </td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>zookeeperSessionTimeoutMs</p></td><td
colspan="1" rowspan="1" class="confluenceTd"> </td><td colspan="1"
rowspan="1" class="confluenceTd"> </td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>zookeeperConnectionTimeoutMs</p></td><td
colspan="1" rowspan="1" class="confluenceTd"> </td><td colspan="1"
rowspan="1" class="confluenceTd"> </td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>zookeeperSyncTimeMs</p></td><td colspan="1"
rowspan="1" class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p><span style="color:
rgb(51,51,51);">consumersCount</span></p></td><td colspan="1" rowspan="1"
class="confl
uenceTd"><p><span style="color: rgb(51,51,51);">1</span></p></td><td
colspan="1" rowspan="1" class="confluenceTd"><p><strong>Camel 2.15.0:</strong>
The number of consumers that connect to kafka server</p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p><span style="color:
rgb(51,51,51);">batchSize</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><span style="color:
rgb(51,51,51);">100</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p><strong>Camel 2.15.0: </strong>The batchSize that the
BatchingConsumerTask processes once, deprecated since <strong>2.17.1</strong>,
removed<strong><br clear="none"></strong>since
<strong>2.18.0</strong></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p><span style="color:
rgb(51,51,51);">barrierAwaitTimeoutMs</span></p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p><span style="color:
rgb(51,51,51);">10000</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>
<strong>Camel 2.15.0: </strong>If the BatchingConsumerTask processes exchange
exceed the batchSize, it will wait for <span style="color:
rgb(51,51,51);">barrierAwaitTimeoutMs</span>, deprecated since
<strong>2.17.1</strong>, removed since <strong>2.18.0</strong><strong>.<br
clear="none"></strong></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">bridgeEndpoint</td><td colspan="1" rowspan="1"
class="confluenceTd">false</td><td colspan="1" rowspan="1"
class="confluenceTd">Camel 2.16.0: If bridgeEndpoint is true, the producer will
ignore the topic header setting of the
message.</td></tr></tbody></table></div></div>
+
+
+<p>You can append query options to the URI in the following format,
<code>?option=value&option=value&...</code></p><h3
id="Kafka-ProducerOptions(Camel2.16orolder)">Producer Options (Camel 2.16
or older)</h3><div class="confluenceTableSmall"><div class="table-wrap"><table
class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh"><p>Property</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Default</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Description</p></th></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>producerType</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p>sync (Taken from native KafkaProducer
class)</p></td><td colspan="1" rowspan="1" class="confluenceTd"><p>sync - send
message/batch immediately, and wait until response is received</p><p>async -
queue the message/batch to send. There is a thread per broker (Kafka node)
which polls from this queue upon <span>queueBufferingMaxMs or <sp
an>batchNumMessages</span></span></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">compressionCodec</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">compressedTopics</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">messageSendMaxRetries</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">retryBackoffMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">topicMetadataRefreshIntervalMs</td><td colspan="1"
rowspan="1" class="confluenceTd"> </td><td
colspan="1" rowspan="1" class="confluenceTd"> </td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd">sendBufferBytes</td><td
colspan="1" rowspan="1" class="confluenceTd"> </td><td colspan="1"
rowspan="1" class="confluenceTd"> </td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd">requestRequiredAcks</td><td colspan="1"
rowspan="1" class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">requestTimeoutMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">queueBufferingMaxMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">queueBufferingMaxMessages</td><td colspan="1" rowspan="1"
class="confluenceTd">&#
160;</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">queueEnqueueTimeoutMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">batchNumMessages</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">serializerClass</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">keySerializerClass</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr></tbody></table></div></div>
+
+
+<h3 id="Kafka-ConsumerOptions(Camel2.16orolder)">Consumer Options (Camel
2.16 or older)</h3><div class="confluenceTableSmall"><div
class="table-wrap"><table class="confluenceTable"><tbody><tr><th colspan="1"
rowspan="1" class="confluenceTh"><p>Property</p></th><th colspan="1"
rowspan="1" class="confluenceTh"><p>Default</p></th><th colspan="1" rowspan="1"
class="confluenceTh"><p>Description</p></th></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>consumerId</p></td><td colspan="1"
rowspan="1" class="confluenceTd"><p> </p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">socketTimeoutMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">socketReceiveBufferBytes</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1" class="conf
luenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">fetchMessageMaxBytes</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">autoCommitEnable</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">autoCommitIntervalMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">queuedMaxMessages</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">rebalanceMaxRetries</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1" c
lass="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">fetchMinBytes</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">fetchWaitMaxMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">rebalanceBackoffMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">refreshLeaderBackoffMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">autoOffsetReset</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1" c
lass="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd">consumerTimeoutMs</td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr></tbody></table></div></div>
+
+
+<h3 id="Kafka-Options(Camel2.17ornewer)">Options (Camel 2.17 or
newer)</h3><div class="table-wrap"><table
class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh">Property</th><th colspan="1" rowspan="1"
class="confluenceTh">Default</th><th colspan="1" rowspan="1"
class="confluenceTh">Description</th></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span class="s1">topic</span></p></td><td
colspan="1" rowspan="1" class="confluenceTd"> </td><td colspan="1"
rowspan="1" class="confluenceTd">Topic to use. From the
<strong>consumer</strong> side you can specify also a comma separated list of
topics.</td></tr><tr><td colspan="1" rowspan="1" class="confluenceTd"><p
class="p1"><span class="s1">groupId</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span class="s1">consumerStreams
</span></p></td><td colspan="1" rowspan="1" class="confluenceTd">10</td><td
colspan="1" rowspan="1" class="confluenceTd"> </td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p class="p1"><span
class="s1">clientId</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd"> </td><td colspan="1" rowspan="1"
class="confluenceTd"> </td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span
class="s1">consumersCount</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd">1</td><td colspan="1" rowspan="1" class="confluenceTd"><p
class="p1"><span class="s1">The number of consumers that connect to kafka
server</span></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span
class="s1">batchSize</span></p></td><td colspan="1" rowspan="1"
class="confluenceTd">100</td><td colspan="1" rowspan="1"
class="confluenceTd"><p class="p1"><span class="s1">Commit Size if auto commit
is false</span></p></td></tr><tr
><td colspan="1" rowspan="1" class="confluenceTd"><p class="p1"><span
>class="s1">bridgeEndpoint</span></p></td><td colspan="1" rowspan="1"
>class="confluenceTd">false</td><td colspan="1" rowspan="1"
>class="confluenceTd"><span>If the bridgeEndpoint is true, the producer will
>ignore the topic header setting of the
>message.</span></td></tr></tbody></table></div><p> </p><h3
>id="Kafka-ProducerOptions(Camel2.17ornewer)">Producer Options (Camel
>2.17 or newer)</h3><div class="table-wrap"><table
>class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
>class="confluenceTh">Property</th><th colspan="1" rowspan="1"
>class="confluenceTh">Default & Description Reference</th></tr><tr><td
>colspan="1" rowspan="1" class="confluenceTd"><p>serializerClass</p></td><td
>colspan="1" rowspan="46" class="confluenceTd"><p class="p1"><span
>class="s1"><span class="nolink"><a shape="rect" class="external-link"
>href="http://kafka.apache.org/documentation.html#producerconfigs";>http://kafka.apache
.org/documentation.html#producerconfigs</a></span></span></p><p
class="p1"><span class="s1"><span class="nolink">serializerClass : <span
class="pl-s">org.apache.kafka.common.serialization.StringSerializer</span></span></span></p><p
class="p1"><span class="s1"><span class="nolink">keySerializerClass : <span
class="s1"><span class="nolink"><span
class="pl-s">org.apache.kafka.common.serialization.StringSerializer</span></span></span></span></span></p><p>partitioner
: <span
class="pl-s">org.apache.kafka.clients.producer.internals.DefaultPartitioner<span
class="pl-pds"> </span></span></p><p> </p><p> </p><p
class="p1"><span class="s1"><span class="nolink"><br
clear="none"></span></span></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>keySerializerClass</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>requestRequiredAcks </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>bufferMemorySize </p></td></tr><tr>
<td colspan="1" rowspan="1"
class="confluenceTd"><p>compressionCodec </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>retries </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeyPassword</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslKeystoreLocation</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeystorePassword</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslTruststoreLocation</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslTruststorePassword</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>producerBatchSize </p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>clientId</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>connectionMaxIdleMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>lingerMs </p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>maxBlockMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>maxRequestSize </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>partitioner </p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>receiveBufferBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>requestTimeoutMs </p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>saslKerberosServiceName</p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>saslMechanism <strong>(from
Camel 2.18)</strong></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>securityProtocol</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sendBufferBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslEnabledProtocols </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeystoreType </p></td></tr><tr><t
d colspan="1" rowspan="1"
class="confluenceTd"><p>sslProtocol </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslProvider</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslTruststoreType</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>maxInFlightRequest </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>metadataMaxAgeMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>metricReporters</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>noOfMetricsSample </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>metricsSampleWindowMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>reconnectBackoffMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>retryBackoffMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>kerberosInitCmd </p></td></tr><tr><td
cols
pan="1" rowspan="1"
class="confluenceTd"><p>kerberosBeforeReloginMinTime </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosRenewJitter </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosRenewWindowFactor </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslCipherSuites</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslEndpointAlgorithm</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeymanagerAlgorithm </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslTrustmanagerAlgorithm </p></td></tr></tbody></table></div><h3
id="Kafka-ConsumerOptions(Camel2.17ornewer)">Consumer Options (Camel 2.17
or newer)</h3><div class="table-wrap"><table
class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh">Property</th><th colspan="1" rowspan="1"
class="confluenceTh">Default & Description Reference</th>
</tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p> </p></td><td colspan="1" rowspan="45"
class="confluenceTd"><p class="p1"><span class="s1"><span class="nolink"><a
shape="rect" class="external-link"
href="http://kafka.apache.org/documentation.html#newconsumerconfigs";>http://kafka.apache.org/documentation.html#newconsumerconfigs</a></span></span></p><p
class="p1"><span class="s1"><span class="nolink">keyDeserializer : <span
class="pl-s">org.apache.kafka.common.serialization.StringDeserializer</span></span></span></p><p
class="p1"><span class="s1"><span class="nolink">valueDeserializer : <span
class="s1"><span class="nolink"><span
class="pl-s">org.apache.kafka.common.serialization.StringDeserializer</span></span></span></span></span></p><p>partitionAssignor
: <span
class="pl-s">org.apache.kafka.clients.consumer.RangeAssignor</span></p><p> </p><p
class="p1"><span class="s1"><span class="nolink"><span class="s1"><span
class="nolink"><span class="pl-s"><br clear="no
ne"></span></span></span></span></span></p><p class="p1"><span
class="s1"><span class="nolink"><br
clear="none"></span></span></p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>keyDeserializer</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>valueDeserializer</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>fetchMinBytes </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>groupId</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>heartbeatIntervalMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>maxPartitionFetchBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sessionTimeoutMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslKeyPassword</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeystoreLocation</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslKe
ystorePassword</p></td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>sslTruststoreLocation</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslTruststorePassword</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>autoOffsetReset </p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>connectionMaxIdleMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>autoCommitEnable </p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>partitionAssignor </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>receiveBufferBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>consumerRequestTimeoutMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>saslKerberosServiceName</p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p><span>saslMechanism
</span><strong>(from Camel 2.18)</strong></p></td></t
r><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>securityProtocol</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sendBufferBytes </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslEnabledProtocols </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslKeystoreType </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslProtocol </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslProvider</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>sslTruststoreType</p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>autoCommitIntervalMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>checkCrcs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>clientId</p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>fetchWaitMaxMs </p></td></tr><tr><td
colspan="1" rowspan=
"1" class="confluenceTd"><p>metadataMaxAgeMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>metricReporters</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>noOfMetricsSample </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>metricsSampleWindowMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>reconnectBackoffMs </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>retryBackoffMs </p></td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><p>kerberosInitCmd </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosBeforeReloginMinTime </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosRenewJitter </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>kerberosRenewWindowFactor </p></td></tr><tr><td
colspan="1" rowspan="1" class="confluenceTd"><p>sslCipherSuites</p></td></t
r><tr><td colspan="1" rowspan="1"
class="confluenceTd"><p>sslEndpointAlgorithm</p></td></tr><tr><td colspan="1"
rowspan="1"
class="confluenceTd"><p>sslKeymanagerAlgorithm </p></td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><p>sslTrustmanagerAlgorithm </p></td></tr></tbody></table></div><p> </p><h3
id="Kafka-Samples">Samples</h3><h4 id="Kafka-Camel2.16orolder.1">Camel 2.16 or
older</h4><p>Consuming messages:</p><div class="code panel pdl"
style="border-width: 1px;"><div class="codeContent panelContent pdl">
<script class="brush: java; gutter: false; theme: Default"
type="syntaxhighlighter"><![CDATA[from("kafka:localhost:9092?topic=test&zookeeperHost=localhost&zookeeperPort=2181&groupId=group1").to("log:input");
]]></script>
</div></div><p>Producing messages:</p><p>See unit tests of camel-kafka for
more examples</p><h4 id="Kafka-Camel2.17ornewer.1">Camel 2.17 or
newer</h4><p>Consuming messages:</p><div class="code panel pdl"
style="border-width: 1px;"><div class="codeContent panelContent pdl">
@@ -149,6 +152,42 @@
}
}).to("kafka:localhost:9092?topic=test");
]]></script>
+</div></div><h3
id="Kafka-UsingtheKafkaidempotentrepository(AvailablefromCamel2.19)">Using the
Kafka idempotent repository (Available from Camel 2.19)</h3><p>The
<code>camel-kafka</code> library provides a Kafka topic-based idempotent
repository. This repository stores broadcasts all changes to idempotent state
(add/remove) in a Kafka topic, and populates a local in-memory cache for each
repository's process instance through event sourcing.</p><p>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.</p><p>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.</p><p>On start
up, 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 <code>pollDurationMs</code> in length
returns 0 records. Startup will not be completed until either the cache has
warmed up, or 30 seconds go by; if the latter happens the idempotent repository
may be in an inconsistent state until its consumer catches up to the end of the
topic.</p><p>A <code>KafkaIdempotentRepository</code> has the following
properties:</p><div class="table-wrap"><table
class="confluenceTable"><tbody><tr><th colspan="1" rowspan="1"
class="confluenceTh">Property</th><th colspan="1" rowspan="1"
class="confluenceTh">Description</th></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><pre><span style="color:
rgb(102,14,122);">topic</span></pre></td><td colspan="1" rowspan="1"
class="confluenceTd">The name of the Kafka topic to use to broadcast changes.
(required)</td></tr><tr><td
colspan="1" rowspan="1"
class="confluenceTd"><code>bootstrapServers</code></td><td colspan="1"
rowspan="1" class="confluenceTd">The <code>bootstrap.servers</code> property on
the internal Kafka producer and consumer. Use this as shorthand if not setting
<code>consumerConfig</code> and <code>producerConfig</code>. If used, this
component will apply sensible default configurations for the producer and
consumer.</td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><code>producerConfig</code></td><td colspan="1"
rowspan="1" class="confluenceTd">Sets the properties that will be used by the
Kafka producer that broadcasts changes. Overrides
<code>bootstrapServers</code>, so must define the Kafka
<code>bootstrap.servers</code> property itself</td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><code>consumerConfig</code></td><td
colspan="1" rowspan="1" class="confluenceTd">Sets the properties that will be
used by the Kafka consumer that populates the cache from the topic.
Overrides <code>bootstrapServers</code>, so must define the Kafka
<code>bootstrap.servers</code> property itself</td></tr><tr><td colspan="1"
rowspan="1" class="confluenceTd"><code>maxCacheSize</code></td><td colspan="1"
rowspan="1" class="confluenceTd">How many of the most recently used keys should
be stored in memory (default 1000).</td></tr><tr><td colspan="1" rowspan="1"
class="confluenceTd"><pre><span style="color:
rgb(0,0,0);">pollDurationMs</span></pre></td><td colspan="1" rowspan="1"
class="confluenceTd"><p>The poll duration of the Kafka consumer. The local
caches are updated immediately; this value will affect how far behind other
peers in the cluster are, which are updating their caches from the topic,
relative to the idempotent consumer instance issued the cache action
message.</p><p>The default value of this is 100 ms. If setting this value
explicitly, be aware that there is a tradeoff between the remote cache liveness
and the volume of network traffic between this repo
sitory's consumer and the Kafka
brokers.</p></td></tr></tbody></table></div><p>The repository can be
instantiated by defining the topic and <code>bootstrapServers</code>, or the
<code>producerConfig</code> and <code>consumerConfig</code> property sets can
be explicitly defined to enable features such as SSL/SASL.</p><p>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 <code>CamelContext</code>
aware.</p><p>Sample usage is as follows:</p><div class="code panel pdl"
style="border-width: 1px;"><div class="codeContent panelContent pdl">
+<script class="brush: java; gutter: false; theme: Default"
type="syntaxhighlighter"><![CDATA[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()]]></script>
+</div></div><p>In XML:</p><div class="code panel pdl" style="border-width:
1px;"><div class="codeContent panelContent pdl">
+<script class="brush: xml; gutter: false; theme: Default"
type="syntaxhighlighter"><![CDATA[<!-- 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>
+]]></script>
</div></div><p> </p><p></p><h2 id="Kafka-Endpoints">Endpoints</h2>
<p></p><p>Camel supports the <a shape="rect"
href="message-endpoint.html">Message Endpoint</a> pattern using the <a
shape="rect" class="external-link"
href="http://camel.apache.org/maven/current/camel-core/apidocs/org/apache/camel/Endpoint.html";>Endpoint</a>
interface. Endpoints are usually created by a <a shape="rect"
href="component.html">Component</a> and Endpoints are usually referred to in
the <a shape="rect" href="dsl.html">DSL</a> via their <a shape="rect"
href="uris.html">URIs</a>. </p>
