INSTRUCTION.txt

Pre-requisites:

 - Maven 3.x
 - Java 1.6 or above
 
Tested Platform: 

 - Microsoft WINDOWS V-7
 - UBUNTU 14.04
 - wso2esb-5.0.0
 - Java 1.8

1. Create Topic and start the zookeeper, kafka server and also producer to send a json/xml message to test the consumption via WSO2 ESB.

2. Navigate to "<INBOUND_HOME>" and run the following command.
        $ mvn clean install

3. To use the Kafka inbound endpoint, you need to download the inbound jar from https://storepreview.wso2.com/store/ and copy the jar to the <ESB_HOME>/repository/components/dropins directory.

4. Configuration:

without security

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"
                 name="kafka"
                 sequence="request"
                 onError="fault"
                 class="org.wso2.carbon.inbound.kafka.KafkaMessageConsumer"
                 suspend="false">
   <parameters>
      <parameter name="sequential">true</parameter>
      <parameter name="interval">10</parameter>
      <parameter name="coordination">true</parameter>
      <parameter name="inbound.behavior">polling</parameter>
      <parameter name="value.deserializer">org.apache.kafka.common.serialization.StringDeserializer</parameter>
      <parameter name="topic.names">test</parameter>
      <parameter name="poll.timeout">100</parameter>
      <parameter name="bootstrap.servers">localhost:9092</parameter>
      <parameter name="group.id">hello</parameter>
      <parameter name="contentType">application/json</parameter>
      <parameter name="key.deserializer">org.apache.kafka.common.serialization.StringDeserializer</parameter>
   </parameters>
</inboundEndpoint>

with Security

<?xml version="1.0" encoding="UTF-8"?>
<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"
                 name="kafka"
                 sequence="request"
                 onError="fault"
                 class="org.wso2.carbon.inbound.kafka.KafkaMessageConsumer"
                 suspend="false">
   <parameters>
      <parameter name="interval">10</parameter>
      <parameter name="coordination">true</parameter>
      <parameter name="sequential">true</parameter>
      <parameter name="inbound.behavior">polling</parameter>
      <parameter name="value.deserializer">org.apache.kafka.common.serialization.StringDeserializer</parameter>
      <parameter name="topic.names">test</parameter>
      <parameter name="poll.timeout">100</parameter>
      <parameter name="bootstrap.servers">localhost:9092</parameter>
      <parameter name="group.id">hello</parameter>
      <parameter name="contentType">application/json</parameter>
      <parameter name="key.deserializer">org.apache.kafka.common.serialization.StringDeserializer</parameter>
      <parameter name="ssl.keystore.location">/home/hariprasath/Desktop/kafkaNewJira/certKafka/kafka.client.keystore.jks</parameter>
      <parameter name="ssl.keystore.password">test1234</parameter>
      <parameter name="security.protocol">SSL</parameter>
      <parameter name="ssl.truststore.location">/home/hariprasath/Desktop/kafkaNewJira/certKafka/kafka.client.truststore.jks</parameter>
      <parameter name="ssl.truststore.password">test1234</parameter>
      <parameter name="contentType">application/json</parameter>
   </parameters>
</inboundEndpoint>