Kafka is a distributed, partitioned, replicated commit log service. It provides the functionality of a messaging system. Kafka maintains feeds of messages in topics. Producers write data to topics and consumers read from topics. For more information on Apache Kafka, go to Apache Kafka documentation.
WSO2 ESB kafka inbound endpoint acts as a message consumer. It creates a connection to ZooKeeper and requests messages for a topic, topics or topic filters.
In order to use the kafka inbound endpoint, you need to download and install Apache Kafka. The recommend version is
To configure the kafka inbound endpoint, copy the following client libraries from the
<KAFKA_HOME>/libs directory to the
- If you are using
kafka_2.x.x-0.8.2.0or later, you also need to add the
kafka-clients-0.8.x.x.jarfile to the
- If you are using a newer version of ZooKeeper, create a directory name
<ESB_HOME>/repository/confdirectory, and then add the
jaas.conffile to the
identitydirectory. This is required because Kerberos authentication is enforced on newer versions of ZooKeeper.
Configuration parameters for a kafka inbound endpoint are XML fragments that represent various properties.
Following is a sample high level kafka configuration that can be used to consume messages using the specified topic or topics:
Kafka inbound endpoint parameters for a high level configuration
|Possible Values||Default Value|
|The host and port of a ZooKeeper server (||Yes||localhost:2181|
|The consumer configuration type. This can either be ||Yes||highlevel, simple|
|The polling interval for the inbound endpoint to poll the messages.||Yes|
|If set to ||Yes||true, false||true|
|The behaviour when executing the given sequence.||Yes||true, false||true|
|The category to feed the messages. A high level kafka configuration can have more than one topic. You can specify multiple topic names as comma separated values.||Yes|
|The content of the message.||Yes||appllication/xml, application/json|
If all the consumer instances have the same consumer group, this works as a traditional queue balancing the load over the consumers.
If all the consumer instances have different consumer groups, this works as publish-subscribe and all messages are broadcast to all consumers.
|The number of threads.||No||1|
|The id of the consumer.||No||null|
|The socket timeout for network requests.||No||30 * 1000|
|The socket receive buffer for network requests.||No||64 * 1024|
|The number of byes of messages to attempt to fetch for each topic-partition in each fetch request.||No||1024 * 1024|
|The number fetcher threads used to fetch data.||No||1|
|The committed offset to be used as the position from which the new consumer will begin when the process fails.||No||true, false||true|
|The frequency in ms that the consumer offsets are committed to zookeeper.||No||60 * 1000|
|The maximum number of message chunks buffered for consumption. Each chunk can go up to the value specified in ||No||2|
|The maximum number of retry attempts.||No||4|
|The minimum amount of data the server should return for a fetch request.||No||1|
|The maximum amount of time the server will block before responding to the fetch request when sufficient data is not available to immediately serve ||No||100|
|The backoff time between retries during rebalance.||No||2000|
|The backoff time to wait before trying to determine the leader of a partition that has just lost its leader.||No||200|
Set this to one of the following values based on what you need to do when there is no initial offset in ZooKeeper or if an offset is out of range.
smallest - Automatically reset the offset to the smallest offset.
|No||smallest, largest, anything else||largest|
|The timeout interval after which a timeout exception is to be thrown to the consumer if no message is available for consumption. It is a good practice to set this value lower than the interval of the Kafka inbound endpoint.||No||3000|
|Set to ||No||true, false||true|
|The partitions assignment strategy to be used when assigning partitions to consumer streams.||No||range, roundrobin||range|
|The user specified string sent in each request to help trace calls.||No||value of group id|
|The ZooKeeper session timeout value in milliseconds.||No||6000|
|The maximum time in milliseconds that the client should wait while establishing a connection to ZooKeeper.||No||6000|
|The time difference in milliseconds that a ZooKeeper follower can be behind a ZooKeeper leader.||No||2000|
|The offsets storage location.||No||zookeeper, kafka||zookeeper|
|The backoff period in milliseconds when reconnecting the offsets channel or retrying failed offset fetch/commit requests.||No||1000|
|The socket timeout in milliseconds when reading responses for offset fetch/commit requests.||No||10000|
|The maximum retry attempts allowed. If a consumer metadata request fails for any reason, retry takes place but does not have an impact on this limit.||No||5|
|If ||No||true, false||true|
Following is a sample high level kafka configuration that can be used to consume messages using the topics filter, which can either be a white list or a black list:
In the above configuration, you will see that the following parameters are set to consume topic filters:
In high level kafka configurations, the follwing parameters are used instead of the
<parameter name="topic.filter">test</parameter> <parameter name="filter.from.whitelist">true</parameter>
Following are descriptions of the parameters set to consume topic filters in a kafka configuration:
|The name of the topic filter.||Yes|
|If this is set to |
If this is set to
Following is a sample low level kafka configuration that can be used to consume messages from a specific server in a specific partition, so that the messages are limited:
Kafka inbound endpoint parameters for a low level configuration
|The category to feed the messages.||Yes|
|The specific Kafka broker name.||Yes|
|The specific Kafka server port number.||Yes|
|The partition of the topic.||Yes|
The maximum number of messages to retrieve.
For a sample that demonstrates how one way message bridging from Kafka to HTTP can be done using the kafka inbound endpoint, see Sample 904: Kafka Inbound Endpoint Sample.