DBCP Idle Connections not being reused in Camel Ro

2020-07-23 04:05发布

站内问答 / 前端开发
1490 1 4

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&amp;from=Error@Camel.com.au&amp;subject=GENERAL ERROR: A File Could Not Be Imported&amp;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&amp;move=.donebackup/${date:now:yyyyMMdd}/${file:onlyname.noext}_DONE_${date:now:yyyyMMddHHmmss}.${file:name.ext}&amp;readLock=changed&amp;include=.*.dl&amp;maxMessagesPerPoll=0&amp;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&amp;from=Error@yadda.com&amp;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&amp;from=Error@yadda.com&amp;subject=ERROR: A File Could Not Be Imported&amp;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:

enter image description here

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.

1条回答
地球回转人心会变
2楼-- · 2020-07-23 05:06

Found the problem. This was a real picnic error or id10t mistake on my side.

In my ETL beans I had the following lines of code

try
 {
        conn= dataSource.getConnection();
        stmt =conn.prepareStatement(SQL_ETL_INSERT);     
        stmt.setString(1, (String)hdr.get("CamelFileName"));
        stmt.executeUpdate();        
        conn= dataSource.getConnection();

        stmt =conn.prepareStatement(SQL_ETL_DELETE);
        stmt.setString(1, (String)hdr.get("CamelFileName"));
        stmt.executeUpdate();    

    }
    catch (Exception e)
    {

        throw new Exception(e.getMessage());

    }

    finally
    {
        try
        {
                if (stmt!=null)
                {
                    stmt.close();
                    stmt= null;
                }
                if (conn!=null)
                {
                    conn.close();
                    conn= null;
                }
        }
        catch(SQLException e)
        {

            throw new Exception(e.getMessage());

        }

    }

Notice I run the conn= dataSource.getConnection(); twice! Then I release only one of the connections. Copy and paste and late nights coding does not mix.

查看更多
0人赞 添加讨论(0) 举报
登录 后发表回答
相关问题
查看全部
相关文章
查看全部
收藏的人(4)