I am pretty sure the idle connections are not being re-used or I am leaking connections. I have a simple route that start from a file consumer. The file consumer consumes text files. After picking up the file I check a table to ensure that this is not a duplicate file.
I then convert the message body from a file to string. I then split the file up and run the individual pieces through a route depending on what type of record it is. Each one of these routes eventually inserts this record into staging table on a server running on MySQL.
Below is a simplified version of the route.
<bean id="myPool" class="java.util.concurrent.Executors" factory-method="newFixedThreadPool">
<argument index="0" value="8"/>
</bean>
<camelContext trace="false" handleFault="true" errorHandlerRef="redeliveryErrorHandler" id="FileETLProcess" xmlns="http://camel.apache.org/schema/blueprint">
<errorHandler type="DefaultErrorHandler" useOriginalMessage="true" id="redeliveryErrorHandler">
<redeliveryPolicy maximumRedeliveries="3" redeliveryDelay="25" backOffMultiplier="2" useExponentialBackOff="false" retryAttemptedLogLevel="INFO"/>
</errorHandler>
<onException useOriginalMessage="true">
<exception>java.lang.Exception</exception>
<handled>
<constant>true</constant>
</handled>
<log message="STARTING ERROR GENERAL HANDLER"/>
<bean ref="GeneralError"/>
<to uri="smtp://mailsrv?to=x.x@yadda.com&from=Error@Camel.com.au&subject=GENERAL ERROR: A File Could Not Be Imported&contentType=text/html"/>
<to uri="file:d:/Inbox/.badFile?fileName=${file:onlyname.noext}_GENERALERROR_${date:now:yyyyMMddHHmmss}.${file:name.ext}"/>
</onException>
<route id="ExtractFileRoute">
<from uri="file:d:/Inbox?delay=10000&move=.donebackup/${date:now:yyyyMMdd}/${file:onlyname.noext}_DONE_${date:now:yyyyMMddHHmmss}.${file:name.ext}&readLock=changed&include=.*.dl&maxMessagesPerPoll=0&sortBy=${file:length}"/>
<bean ref="FileCheck"/>
<choice>
<when>
<simple>${header.ACCEPTEDFILE} == 'YES'</simple>
<log message="File Import Route Started At:${date:now:yyyy-MM-dd HH:mm:ss}"/>
<convertBodyTo type="java.lang.String"/>
<log message="Converted File To String:${date:now:yyyy-MM-dd HH:mm:ss} handing data to File To DB route."/>
<split parallelProcessing="true" executorServiceRef="myPool" streaming="true" shareUnitOfWork="true">
<tokenize token="\n"></tokenize>
<setHeader headerName="SPLITFINISHED">
<simple>${property.CamelSplitComplete}</simple>
</setHeader>
<setHeader headerName="SPLITNUMBER">
<simple>${property.CamelSplitIndex}</simple>
</setHeader>
<bean ref="EnrichHeader"/>
<choice>
<when>
<simple>${header.RECORDTYPE} == 'HEADER'</simple>
<doTry>
<unmarshal ref="bindyHeader"/>
<bean ref="HeaderPersist"/>
<choice>
<when>
<simple>${property.CamelSplitComplete} == true</simple>
<to uri="direct:auxrecordsmove"/>
</when>
</choice>
<doCatch>
<exception>java.lang.Exception</exception>
<handled>
<constant>true</constant>
</handled>
<bean ref="RecordErrorReport"/>
<choice>
<when>
<simple>${property.CamelSplitComplete} == true</simple>
<to uri="direct:auxrecordsmove"/>
</when>
</choice>
</doCatch>
</doTry>
</when>
<when>
<simple>${header.RECORDTYPE} == 'A'</simple>
<doTry>
<unmarshal ref="bindyAccount"/>
<bean ref="AccountPersist"/>
<choice>
<when>
<simple>${property.CamelSplitComplete} == true</simple>
<to uri="direct:auxrecordsmove"/>
</when>
</choice>
<doCatch>
<exception>java.lang.Exception</exception>
<handled>
<constant>true</constant>
</handled>
<bean ref="RecordErrorReport"/>
<choice>
<when>
<simple>${property.CamelSplitComplete} == true</simple>
<to uri="direct:auxrecordsmove"/>
</when>
</choice>
</doCatch>
</doTry>
</when>
<when>
<simple>${header.RECORDTYPE} == 'C'</simple>
<doTry>
<unmarshal ref="bindyComaker"/>
<bean ref="CoMakerPersist"/>
<choice>
<when>
<simple>${property.CamelSplitComplete} == true</simple>
<to uri="direct:auxrecordsmove"/>
</when>
</choice>
<doCatch>
<exception>java.lang.Exception</exception>
<handled>
<constant>true</constant>
</handled>
<bean ref="RecordErrorReport"/>
<choice>
<when>
<simple>${property.CamelSplitComplete} == true</simple>
<to uri="direct:auxrecordsmove"/>
</when>
</choice>
</doCatch>
</doTry>
</when>
Some other beans here........
<when>
<simple>${property.CamelSplitComplete} == true</simple>
<to uri="direct:auxrecordsmove"/>
</when>
<otherwise>
<to uri="smtp://ims-mail?to=ops@ipanic&from=Error@yadda.com&subject=URGENT:UNKOWN RECORD TYPE FOUND IN FILE"/>
<choice>
<when>
<simple>${property.CamelSplitComplete} == true</simple>
<to uri="direct:auxrecordsmove"/>
</when>
</choice>
</otherwise>
</choice>
</split>
</when>
<otherwise>
<to uri="file:d:/RMSInbox/.badFile?fileName=${file:onlyname.noext}_POSSIBLE_DUPLICATE_ERROR_${date:now:yyyyMMddHHmmss}.${file:name.ext}"/>
<bean ref="FileErrorReport"/>
<to uri="smtp://ims-mail?to=ops@panic.com&from=Error@yadda.com&subject=ERROR: A File Could Not Be Imported&contentType=text/html"/>
</otherwise>
</choice>
</route>
So each message on this route eventually hits a bean which will insert it into a database. So I added DBCP to the dependencies and then declare it in my osgi xml blueprint as follows:
<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://Canttouchthis:3306/ETLDB"/>
<property name="username" value="ETLUser"/>
<property name="password" value="password"/>
<property name="initialSize" value="2"/>
<property name="maxActive" value="16"/>
<property name="maxIdle" value="16"/>
<property name="minIdle" value="2"/>
<property name="timeBetweenEvictionRunsMillis" value="180000"/>
<property name="minEvictableIdleTimeMillis" value="180000"/>
<property name="testOnBorrow" value="true"/>
<property name="testWhileIdle" value="true"/>
<property name="testOnReturn" value="true"/>
<property name="validationQuery" value="SELECT 1"/>
<property name="maxWait" value="10000"/>
<property name="removeAbandoned" value="true"/>
<property name="logAbandoned" value="false"/>
<property name="removeAbandonedTimeout" value="300"/>
</bean>
I also declare my beans that will do the processing like this:
<bean id="AccountPersist" class="com.foo.NewAccount.AccountInformationToDatabase">
<property name="dataSource" ref="dataSource"/>
</bean>
Now when the split on the file has been finished I want to ensure that the records match up. Basically the file has account records and some auxiliary information. So I check in the routes when the split is finished and then once the file is completely in the staging tables I run some additional sanity checks in MySQL.
This second route looks something like this:
<route trace="false" id="MoveMatchingAuxRecordsFromStage">
<from uri="direct:auxrecordsmove"/>
<log message="File Import Route Ended At:${date:now:yyyy-MM-dd HH:mm:ss}"/>
<log message="ETL Route Start AT: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<log message="Moving Matching Comaker records at: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<bean ref="CoMakerETL"/>
<log message="Matching Comaker records move finised at: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<log message="Moving Matching Credit History records at: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<bean ref="CreditHistoryETL"/>
<log message="Matching Credit History records move finised at: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<log message="Moving Matching Extra Detail records at: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<bean ref="ExtraDetailInformationETL"/>
<log message="Matching Extra Detail records move finised at: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<log message="Moving Legal Information records at: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<bean ref="LegalInformationETL"/>
<log message="Matching Legal Information records move finised at: ${date:now:yyyy-MM-dd HH:mm:ss}"/>
<log message="ETL Route Finished At ${date:now:yyyy-MM-dd HH:mm:ss}"/>
</route>
So in my testing everything went fine I could import a file quiet effectively like this. My problems started when I placed more than 5 files in the folder. Basically I watch MySQL grow the connection pool to the maximum size and then not re-use the connections.
So we hit 16 concurrent connections they go to sleep after the a couple of files has been loaded then somewhere at the 4,5,6 file all of a sudden I get the following error:
Cannot get a connection, pool error Timeout waiting for idle object
Or as it appears in the log
[ pool-3-thread-35] oveMatchingAuxRecordsFromStage INFO Matching Extra Detail records move finised at: 2013-07-26 17:41:59
[ pool-3-thread-35] oveMatchingAuxRecordsFromStage INFO Moving Legal Information records at: 2013-07-26 17:41:59
[ pool-3-thread-35] DefaultErrorHandler INFO Failed delivery for (MessageId: ID-IMS-WS2013-001-52799-1374824474993-0-2693 on ExchangeId: ID-IMS-WS2013- 001-52799-1374824474993-0-3230). On delivery attempt: 0 caught: java.lang.Exception: Cannot get a connection, pool error Timeout waiting for idle object
[thread #0 - file://d:/RMSInbox] ExtractRMSNewAccountFileRoute INFO STARTING ERROR GENERAL HANDLER
The MySQL max connections have been pushed to a massive 512. I have tried various pool sizes, threading options etc.
As a matter of interest all of my JDBC code follows this structure. It is not complicated SQL just insert statements....
public class RecordParserErrorReporter {
private static final String SQL_INSERT="INSERT INTO `ETL`.`ETLLog` "+
" ( "+
" `ETL_log_text`, "+
" `filename`, "+
" `record_number`, "+
" `error_message`) "+
" VALUES "+
" ( "+
" ?, "+
" ?, "+
" ?, "+
" ? "+
" ); ";
private BasicDataSource dataSource;
public BasicDataSource getDataSource() {
return dataSource;
}
public void setDataSource(BasicDataSource dataSource) {
this.dataSource = dataSource;
}
public void HandleError
(
@Body String msgBody
, @Headers Map hdr
, Exchange exch
)
{
Connection conn = null;
PreparedStatement stmt=null;
try
{
Exception e = exch.getProperty(Exchange.EXCEPTION_CAUGHT,Exception.class);
conn= dataSource.getConnection();
stmt =conn.prepareStatement(SQL_INSERT);
stmt.setString(1, msgBody);
stmt.setString(2, (String)hdr.get("CamelFileName"));
stmt.setInt(3, (Integer)hdr.get("SPLITNUMBER"));
stmt.setString(4, e.getMessage());
stmt.executeUpdate();
}
catch (Exception e)
{
System.out.println(e.getMessage());
}
finally
{
try
{
if (stmt!=null)
{
stmt.close();
}
if (conn!=null)
{
conn.close();
conn= null;
}
}
catch(SQLException e)
{
System.out.println(e.getMessage());
}
}
}
}
How do I track down why my connections are not being reused? Or if I am leaking connections how do I track that down?
Update:
I set the file consumer to consume 1 file on each poll with a second delay between the polls. I can see it creating a new connection pool for each time the route starts and then not reusing the previously created pool. Appears that for every file I queue a new connection pool is created. Not exactly what I want. Is this correct.
I have included a screen shot of how this looks see below:
Update 2:
This not running under spring DM but a OSGI blue print. I am pretty convinced that the datasource is getting instantiated more than once.
Update 3:
Well ok so the datasource is not getting instantiated more than once. The idle connections are simply not used. I did find something interesting though so I suspect this might be related to what I am see more info in the link here: http://fusesource.com/forums/thread.jspa?threadID=4659&tstart=15
Judging by the amount of view this question is getting I am pretty much stuck out there.
https://stackoverflow.com/questions/17875768
italiano
english
français
española
中国
日本の
العربية
Deutsch
한국어
Português
Russian