WSO2 ESB Kafka Inbound and Kafka Connector Performance tuning

In this post, I am going to explain about Kafka connector and Kafka inbound endpoint performance with WSO2 ESB 4.9 and how to tune the performance for producer and consumer use cases. The performance test of Kafka inbound endpoint and Kafka connector was carried out on below machine configuration:

Performance Test Environment :
OS: Ubuntu 64-bit
Memory: 16GiB
Processor: Intel Core i7-4800MQ CPU 2.70GHzx8

ESB Kafka Connector
WSO2 ESB Kafka connector allows you to send the data to Kafka message broker.  

Threads	Messages per a Thread	No of Messages	A Message Size(byte)	Throughput (/s)
10	100	1,000	106	831
10	500	5,000	106	1590
20	500	10,000	106	2251
100	250	25,000	106	2940
1,000	1500	1,500,000	106	2780

Average Throughput(/s) - 2767

You can get highest throughput with default ESB configuration.

ESB Kafka Inbound Endpoint

WSO2 ESB Kafka Inbound endpoint allows you to consume the data from Kafka message broker. 

https://docs.wso2.com/display/ESB490/Working+with+Inbound+Endpoints


Above document includes inbound endpoint common parameters and how to change them for best performance. Even though I provide some recommended values with the performance tuning stats.

You can tune the Kafka inbound endpoint performance by changing the inbound thread pool values. The values can be changed in the location.

<ESB_HOME>/repository/conf/synapse.properties file

On  Sequential Mode

The following performance stat is measured with default inbound thread pool values and sequential mode.

Threads	Messages per a Thread	No of Messages	A Message Size(byte)	Throughput (/s)
10	100	1,000	106	16
10	500	5,000	106	16
20	500	10,000	106	15

Average Throughput(/s) - 15

On Non Sequential Mode

The following performance stat is measured with default inbound thread pool values and non sequential mode.

inbound.threads.core = 20
inbound.threads.max = 100  

Threads	Messages per a Thread	No of Messages	A Message Size(byte)	Throughput (/s)
10	100	1,000	106	150
10	500	5,000	106	151
20	500	10,000	106	126
100	250	25,000	106	136

We have gone through the performance test with default inbound thread pool.
Average Throughput(/s) - 135

Now we have increased the thread pool values and gone through the performance test.

inbound.threads.core = 200
inbound.threads.max = 1000

Threads	Messages per a Thread	No of Messages	A Message Size(byte)	Throughput (/s)
100	100	10,000	106	769
1000	1000	1,000,000	106	800

We can get the throughput around 800 with above thread pool values. 
Average Throughput(/s) - 799

Now we have increased the inbound thread pool values further. The performance is measured with below inbound thread pool values and non sequential mode. The throughput is 711. There is no significant improvement compared with previous throughput.

inbound.threads.core = 500
inbound.threads.max = 2000

Threads	Messages per a Thread	No of Messages	A Message Size(byte)	Throughput (/s)
100	100	10,000	106	757
1,000	1000	1,000,000	106	710

Average Throughput(/s) - 711

You can configure the inbound common parameters according to this performance stats and see massive performance improvements with the inbound thread pool. To get highest performance, I recommend you to use the following inbound thread pool values with non sequential mode.

inbound.threads.core = 200
inbound.threads.max = 1000

WSO2 IS authenticator - LinkedIn

1) Create a new LinkedIn app

2) The app will be assigned an Client ID and Client Secret. In the panel, make sure to set Authorized Redirect URLs to https://localhost:9443/commonauth

3) Build the sso sample from product-is/modules/samples/sso/sso-agent-sample to get the travelocity war file, add the war file to a web server (apache tomcat) and start the web server.

4) Download the IS ditribution and the service pack from http://wso2.com/products/identity-server/ and apply the service pack to the IS pack.

https://docs.wso2.com/display/IS500/Installing+the+Service+Pack

5) Clone the linkedin authentication connector source from https://github.com/katheesR/is-connectors/tree/master/linkedin, build the linkedin connector and copy the jar to the IS_HOME/components/dropins

6) Create a Identity provider from IS management console.

Now you can see the linkedIn configuration under Federated authentications section. Enable and fill the value for client ID, client secret and callback URL which can be got from the step1.

7) Create a service provider from IS management console.

Assertion consumer URL - http://localhost:8081/travelocity.com/home.jsp

8) Extract the certificate from browser by navigating to https://www.linkedin.com/ and place the certificate file in following locations.

IS_HOME/repository/resources/security

Navigate to the above location from command prompt and execute

'keytool -importcert -file CERT_FILE_NAME -keystore client-truststore.jks -alias "LinkedIn"' in command line to import linkedin certificate into keystore. Give "wso2carbon" as password.

import the following two certificates.

keytool -importcert -file www.linkedin.com -keystore client-truststore.jks -alias "linkedin"

keytool -importcert -file DigiCertSHA2SecureServerCA -keystore client-truststore.jks -alias "Dig"

9) go to the travelocity app http://localhost:8081/travelocity.com/index.jsp and click SAML2 redirect login

9) The page will be redirected to the linkedin authentication page.

If the credential is success then you will get the  linkedin account details.

sample LinkedIn authentication connector code - https://github.com/katheesR/is-connectors/tree/master/linkedin

Apache Kafka Qiuckstart

Introduction

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 the topics. One of the use case of Kafka is messaging system. Kafka offers queuing and publish-subscribe model.

Kafka command line tool

• Download Apache Kafka distribution form here 

   

• Extract and goto Kafka home  

    

• Start the Zookeeper   

         bin/zookeeper-server-start.sh config/zookeeper.properties       

• Start the Kafka server   

          bin/kafka-server-start.sh config/server.properties

• Create a topic,

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

• Run a producer and type the message.

bin/kafka-console-producer.sh --broker-list localhost:9092 

-- topic test 

         This is a message
         This is another message 

• Run the consumer,the messages appear in the consumer

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test  --from-beginning

This is a message

This is another message

Kafka Multi Broker Cluster Setup

We will create 3 Kafka brokers (broker0, broker1 and broker2) whose configurations are based on the default.

First we make a configuration file for each of the brokers:

The default broker0 server properties file is.

config/server-1.properties:
    broker.id=0
    port=9092
    log.dir=/tmp/kafka-logs-0

copy the server.properties file for the broker1 and broker2.

> cp config/server.properties config/server-1.properties
> cp config/server.properties config/server-2.properties

Edit these new files and set the following properties:

config/server-1.properties:
    broker.id=1
    port=9093
    log.dir=/tmp/kafka-logs-1

config/server-2.properties:
    broker.id=2
    port=9094
    log.dir=/tmp/kafka-logs-2

Now we have created 3 Kafka  broker cluster. Start the Kafka server with the  appropriate server properties file.

Broker0

> bin/kafka-server-start.sh config/server.properties

Broker1

> bin/kafka-server-start.sh config/server1.properties

Broker2

> bin/kafka-server-start.sh config/server2.properties

WSO2 ESB 4.9.0 - Kafka Support

The WSO2 ESB Kafka Inbound consumes the message from the Kafka brokers. It allows the message consuming at different speed(polling interval), tenant loading and coordination supports.

You can download latest ESB version from http://wso2.com/products/enterprise-service-bus/

Kafka Inbound Use Cases

1. ESB Kafka Inbound as Queue
2. ESB Kafka Inbound as Topic
3. ESB Kafka Inbound consumes from beginning
4. ESB Kafka Inbound consumes from multiple topics
5. ESB Kafka Inbound consume from specific server and topic partition

USE CASE 1 : WSO2 ESB Kafka Inbound as Queue

If the consumer instances are in a same consumer group, then this works like traditional queue. So only one Kafka inbound will consume the message.

Consider the following Kafka inbound configurations (KafkaInboundEP1 and KafkaInboundEP2). Since the group.id parameter values in both inbound configurations are same, both are in a same consumer group. In such case, any one of these endpoints will consumes the message from the topic.

  

KafkaInboundEP1

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"

                name="KafkaInboundEP1"

                sequence="requestHandlerSeq"

                onError="inFaulte"

                protocol="kafka"

                suspend="false">

  <parameters>

     <parameter name="interval">100</parameter>

     <parameter name="coordination">true</parameter>

     <parameter name="sequential">true</parameter>

     <parameter name="zookeeper.connect">localhost:2181</parameter>

     <parameter name="consumer.type">highlevel</parameter>

     <parameter name="content.type">application/xml</parameter>

     <parameter name="topics">topic</parameter>

     <parameter name="group.id">consumer-group</parameter>

  </parameters>

</inboundEndpoint>

KafkaInboundEP2

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"

                name="KafkaInboundEP2"

                sequence="requestHandlerSeq"

                onError="inFaulte"

                protocol="kafka"

                suspend="false">

  <parameters>

     <parameter name="interval">100</parameter>

     <parameter name="coordination">true</parameter>

     <parameter name="sequential">true</parameter>

     <parameter name="zookeeper.connect">localhost:2181</parameter>

     <parameter name="consumer.type">highlevel</parameter>

     <parameter name="content.type">application/xml</parameter>

     <parameter name="topics">topic</parameter>

     <parameter name="group.id">consumer-group</parameter>

  </parameters>

</inboundEndpoint>

USE CASE 2 : WSO2 ESB Kafka Inbound as Topic

If the consumer instances are in different consumer groups, then this works like publish-subscribe and all messages will be broadcasted to all consumers.

In the following Kafka inbound configurations,  both Kafka inbound endpoints (KafkaListenerEP and KafkaListenerEP2) are in different consumer groups since the group.id parameter values are different. So both inbound endpoint can consume the messages from the topic1.

KafkaInboundEP1

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"

                name="KafkaInboundEP2"

                sequence="requestHandlerSeq"

                onError="inFaulte"

                protocol="kafka"

                suspend="false">

  <parameters>

     <parameter name="interval">100</parameter>

     <parameter name="coordination">true</parameter>

     <parameter name="sequential">true</parameter>

     <parameter name="zookeeper.connect">localhost:2181</parameter>

     <parameter name="consumer.type">highlevel</parameter>

     <parameter name="content.type">application/xml</parameter>

     <parameter name="topics">topic1</parameter>

     <parameter name="group.id">test-group1</parameter>

  </parameters>

</inboundEndpoint>

KafkaInboundEP2

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"

                name="KafkaInboundEP2"

                sequence="requestHandlerSeq"

                onError="inFaulte"

                protocol="kafka"

                suspend="false">

  <parameters>

     <parameter name="interval">100</parameter>

     <parameter name="coordination">true</parameter>

     <parameter name="sequential">true</parameter>

     <parameter name="zookeeper.connect">localhost:2181</parameter>

     <parameter name="consumer.type">highlevel</parameter>

     <parameter name="content.type">application/xml</parameter>

     <parameter name="topics">test</parameter>

     <parameter name="group.id">test-group2</parameter>

  </parameters>

</inboundEndpoint>

USE CASE 3 : WSO2 ESB Kafka inbound as message consumer from beginning

The Kafka inbound endpoint allows to consume the messages from beginning. 

The following configuration can be used for this use case. KafkaInboundEP1 is used to consume the message from specific topic partition 1 from server1 and KafkaaInboundEP2 is used to consume the message from specific topic partition 2 from server1. So we can consume  all messages from specific server from beginning.

KafkaInboundEP1

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"

                name="KafkaInboundEP1"

                sequence="requestHandlerSeq"

                onError="inFaulte"

                protocol="kafka"

                interval="1000"

                suspend="false">

  <parameters>   

     <parameter name="zookeeper.connect">localhost:2181</parameter>

     <parameter name="group.id">test-group</parameter>  

     <parameter name="content.type">application/xml</parameter>

     <parameter name="consumer.type">simple</parameter>

     <parameter name="simple.max.messages.to.read">5</parameter>

     <parameter name="simple.topic">topic</parameter>

     <parameter name="simple.brokers">localhost</parameter>

     <parameter name="simple.port">9092</parameter>

     <parameter name="simple.partition">1</parameter>

     <parameter name="interval">100000</parameter>

  </parameters>

</inboundEndpoint>

KafkaInboundEP2

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"

                name="KafkaInboundEP2"

                sequence="requestHandlerSeq"

                onError="inFaulte"

                protocol="kafka"

                interval="1000"

                suspend="false">

  <parameters>   

     <parameter name="zookeeper.connect">localhost:2181</parameter>

     <parameter name="group.id">test-group</parameter>  

     <parameter name="content.type">application/xml</parameter>

     <parameter name="consumer.type">simple</parameter>

     <parameter name="simple.max.messages.to.read">100000</parameter>

     <parameter name="simple.topic">topic</parameter>

     <parameter name="simple.brokers">localhost</parameter>

     <parameter name="simple.port">9092</parameter>

     <parameter name="simple.partition">2</parameter>

     <parameter name="interval">1000</parameter>

  </parameters>

</inboundEndpoint>

USE CASE 4 : A WSO2 ESB Kafka Inbound consume from more than one topic

A Kafka inbound can consume the messages from more than one topics.

Each topics are added using comma separator in the Kafka inbound endpoint configuration. KafkaInboundEP1 can consume the messages from topic1 and topic2.

KafkaInboundEP1

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"

                name="KafkaInboundEP1"

                sequence="requestHandlerSeq"

                onError="inFaulte"

                protocol="kafka"

                suspend="false">

  <parameters>

     <parameter name="interval">100</parameter>

     <parameter name="coordination">true</parameter>

     <parameter name="sequential">true</parameter>

     <parameter name="zookeeper.connect">localhost:2181</parameter>

     <parameter name="consumer.type">highlevel</parameter>

     <parameter name="content.type">application/xml</parameter>

     <parameter name="topics">topic1,topic2</parameter>

     <parameter name="group.id">test-group</parameter>

  </parameters>

</inboundEndpoint>

USE CASE 5 : WSO2 ESB Kafka Inbound as consume the message from specific server and specific partition

The Kafka Inbound allows to consume the messages from specific Kafka server and specific topic partition.

In the following configuration, The messages are consumed from Kafka server localhost:9092 and topic partition 1.

KafkaInboundEP

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"

                name="KafkaInboundEP2"

                sequence="requestHandlerSeq"

                onError="inFaulte"

                protocol="kafka"

                interval="1000"

                suspend="false">

  <parameters>   

     <parameter name="zookeeper.connect">localhost:2181</parameter>

     <parameter name="group.id">test-group</parameter>  

     <parameter name="content.type">application/xml</parameter>

     <parameter name="consumer.type">simple</parameter>

     <parameter name="simple.max.messages.to.read">100000</parameter>

     <parameter name="simple.topic">topic</parameter>

     <parameter name="simple.brokers">localhost</parameter>

     <parameter name="simple.port">9092</parameter>

     <parameter name="simple.partition">1</parameter>

     <parameter name="interval">1000</parameter>

  </parameters>

</inboundEndpoint>