Broken pipe exception when connecting to cassandra

 Recently I required to WSO2 BAM receiver with high load. And during that I experienced the below exception.

[2013-03-04 15:56:43,010] ERROR {me.prettyprint.cassandra.connection.client.HThriftClient} -  Could not flush transport (to be expected if the pool is shutting down) in close for client: CassandraClient<cassandra03:9170-1581>
org.apache.thrift.transport.TTransportException: java.net.SocketException: Broken pipe
    at org.apache.thrift.transport.TIOStreamTransport.write(TIOStreamTransport.java:147)
    at org.apache.thrift.transport.TFramedTransport.flush(TFramedTransport.java:156)
    at me.prettyprint.cassandra.connection.client.HThriftClient.close(HThriftClient.java:98)
    at me.prettyprint.cassandra.connection.client.HThriftClient.close(HThriftClient.java:26)
    at me.prettyprint.cassandra.connection.HConnectionManager.closeClient(HConnectionManager.java:323)
    at me.prettyprint.cassandra.connection.HConnectionManager.operateWithFailover(HConnectionManager.java:272)
    at me.prettyprint.cassandra.model.ExecutingKeyspace.doExecuteOperation(ExecutingKeyspace.java:97)
    at me.prettyprint.cassandra.model.MutatorImpl.execute(MutatorImpl.java:243)
    at org.wso2.carbon.databridge.persistence.cassandra.datastore.CassandraConnector.commit(CassandraConnector.java:177)
    at org.wso2.carbon.databridge.persistence.cassandra.datastore.CassandraConnector.insertEventList(CassandraConnector.java:402)
    at org.wso2.carbon.databridge.datasink.cassandra.subscriber.BAMEventSubscriber.receive(BAMEventSubscriber.java:50)
    at org.wso2.carbon.databridge.core.internal.queue.QueueWorker.run(QueueWorker.java:80)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:439)
    at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
    at java.util.concurrent.FutureTask.run(FutureTask.java:138)
    at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
    at java.lang.Thread.run(Thread.java:662)
Caused by: java.net.SocketException: Broken pipe
    at java.net.SocketOutputStream.socketWrite0(Native Method)
    at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:92)
    at java.net.SocketOutputStream.write(SocketOutputStream.java:136)
    at org.apache.thrift.transport.TIOStreamTransport.write(TIOStreamTransport.java:145)
    ... 17 more
[2013-03-04 15:56:43,011] ERROR {me.prettyprint.cassandra.connection.HConnectionManager} -  MARK HOST AS DOWN TRIGGERED for host cassandra03(10.157.4.137):9170
[2013-03-04 15:56:43,011] ERROR {me.prettyprint.cassandra.connection.HConnectionManager} -  Pool state on shutdown: <ConcurrentCassandraClientPoolByHost>:


After a doing more research on the cassandra.yaml configuration, i found out changing the below propeties:

thrift_framed_transport_size_in_mb: 15
thrift_max_message_length_in_mb: 16


Increasing the above parameters will solve the problem, ie, you can increase 'thrift_max_message_length_in_mb' to 64 and 'thrift_framed_transport_size_in_mb' to 60, which will help to get rid of the mentioned exception.

Load Balancing Data Publishers and Sending Events to Multiple Receivers

WSO2 BAM/CEP has hight performance thrift based event receiving model, which basically receives the events via TCP. Even though thrift is high performance receiving protocol, load balancing the thrift events is problematic as you need tcp based load balancer rather http based load balancer. Therefore in WSO2 we have added support to have load balancing between Data bridge receivers (ie, WSO2 BAM servers, CEP servers) from the client side, by sending the events in a round robin manner to BAM servers, such that load of events will be balanced between them.

For this we have added a Wrapper class called LoadBalancingDataPublisher, which uses the AsyncDataPublisher in it. It not only load balances the events between the set of servers and also can send same events to some servers. All the capabilities of using the LoadBalancingDataPublisher is provided in BAM 2.2.0 documentation here, which explains the use cases of using the load balancing data publisher.

This provides more fail over handling also with load balancing, which can detect the node failure and stop further publishing for the dead node and it also recognozes the node startup and it starts load balancing the events from that instance.

I'll provide a more detailed description about using LoadDataPublisher to publish events to BAM/CEP in next article.

Whats is WSO2 BAM 2.0..

WSO2 BAM is a fully open source Business Activity Monitoring. Recently WSO2 BAM 2.0 has been released which a complete revamp of WSO2 BAM 1.x.

It's targeted to monitor not-only WSO2 servers but also other third party servers and applications and Key performance Indicators (KPI) and also providing high scalable storage of data via Cassandra and high business data analytics capability through Apache Hive & Hadoop. Also it has very high performance of data capture by using thrift. WSO2 BAM is very scalable, such as you can scale individual components of BAM according to your requirement. If you need high performance for data storage you can only scale the Cassandra storage as multi-node cluster, and if you target to do more analytics in your captured data then you can scale and have a hadoop cluster which makes your analytics faster. WSO2 BAM also has attractive pluggable dashboards, where some of them are in-built by default and some of them are released by separate toolboxes.WSO2 BAM enables you to not only use the default dashboards but also you can write your own gadgets or visualizing engines to plug and view the data.

The below diagram shows the architecture of WSO BAM.

1. Event/Data Receiving
   You can publish any data to BAM via data agents. Already there are data agents available for WSO2 Servers in order to publish the Service Statistics Monitoring, and WSO2 ESB for Mediation Statistics Monitoring and Activity Mediation Monitoring. Similar to this you write a simple java application to publish your third-party events through thrift as described in http://wso2.org/library/articles/2012/07/creating-custom-agents-publish-events-bamcep. And also you can publish the events via REST api to BAM as described in http://docs.wso2.org/wiki/display/BAM200/REST+API.
   
   
2. Data Dumping in Cassandra
   After receiving the events/data in BAM, it immediately flush the data Cassandra database. Cassandra data base provides high scalability and it's very suitable to store huge amount of data with hight performance.
   
   
3. Running business analytics
   Analyzer framework consists of Apache Hive and Hadoop. Hive is a data warehouse system for Hadoop that facilitates easy data summarization, ad-hoc queries, and the analysis of large datasets stored in Hadoop compatible file systems. Hive provides a mechanism to project structure onto this data and query the data using a SQL-like language called HiveQL. At the same time this language also allows traditional map/reduce programmers to plug in their custom mappers and reducers when it is inconvenient or inefficient to express this logic in HiveQL.

You can access the captured data which is stored in Cassandra and run the analytics on those data via writing the hive scripts. You can find sample hive scrips on WSO BAM samples in http://docs.wso2.org/wiki/display/BAM200/Samples. and more hive related queries can be found in https://cwiki.apache.org/confluence/display/Hive/Tutorial.

4. Storing the result in JDBC storage
   After running the business analytics in you captured big data, you can store the summarized results in JDBC storage such as MySQL, H2 Database, Oracle, etc. This facilitates to plug any external reporting engine or preferred dashboard gadgets. In WSO2 BAM samples we have written the summarized data into H2 database as it's the default database used. You can find more details of writing the results in the JDBC storage in http://kasunweranga.blogspot.com/2012/06/jdbc-storage-handler-for-hive.html.

4. Visualization
   Visualization of the data done by WSO2 Jaggery Framework . This data also can be visualized via writing google gadgets also. You can see the sample gadgets created in WSO2 KPI Monitoring sample in http://docs.wso2.org/wiki/display/BAM200/KPI+Monitoring+Sample. And also there are dashboards which includes specially designed gadgets for Service statistics Monitoring and WSO2 ESB montitoring. 

There is already available reporting using the jasper reporting available in http://docs.wso2.org/wiki/display/BAM200/Reports.

These all phases of WSO2 BAM provides to privilege of monitoring your own data and visualize your business aspects.

Try out WSO2 BAM now http://wso2.com/products/business-activity-monitor/

Enjoy WSO2 BAM!! :)