Java Service Task and HttpClient

[martinfranz]

Hi all,

I try to write a custom java service task which uses one http client to make one call per service task to an external custom api.

PoolingHttpClientConnectionManager > CloseableHTTPClient > Java Service task (async) > 100 * HttpPost / HttpGet

The problem with this is: When I execute more than about 20 of these service tasks at the same time (Timer Event) activiti "crashes". I don't see any errors but i am not able to login again. The activiti explorer is like "freezed". The loading indicator turns red and never stops spinning.
From that point i need to delete the database and install activiti again.

I stack with this problem for many days know and I don't have any idea what could be the problem.

What I have tested so far:
Started 100 threads in a small unit test using a pooled connection manager in order to call the target api 100 times concurrently - no problems on client an server side. Where the client was my mac osx java runtime. No tomcat around.


Maybe you have any ideas? Thanks a lot 🙂

This is my setup for the service task:

activiti-context.xml

// relevant parts

     <!– Application Context Provider –>
     <bean id="applicationContextProvider"
        class="com.sw.activiti.spring.context.ApplicationContextProvider">
     </bean>

          <!– A service provider which holdes the base url to the external api –>
     <bean id="fm3RestServiceProvider"
        class="com.sw.activiti.service.tasks.rest.provider.FM3RestServiceProviderImpl">
           <property name="serviceUrl" value="http://myurl/api/" />
     </bean>

       <!– A custom pooled http client usded in the java service task –>
       <bean id="pooledHttpClient" class="com.sw.activiti.service.http.client.PooledHttpClient">
         <property name="maxTotal" value="1000" />
         <property name="defaultMaxPerRoute" value="50" />
       </bean>
‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍

com.sw.activiti.service.http.client.PooledHttpClient

public class PooledHttpClient implements InitializingBean {

   private int maxTotal;
   
   private int defaultMaxPerRoute;
   
   
   
   public int getMaxTotal() {
      return maxTotal;
   }

   public void setMaxTotal(int maxTotal) {
      this.maxTotal = maxTotal;
   }

   public int getDefaultMaxPerRoute() {
      return defaultMaxPerRoute;
   }

   public void setDefaultMaxPerRoute(int defaultMaxPerRoute) {
      this.defaultMaxPerRoute = defaultMaxPerRoute;
   }


   private PoolingHttpClientConnectionManager connectionManager;
   
   private CloseableHttpClient httpclient;
   
   
   public PoolingHttpClientConnectionManager getConnectionManager() {
      return connectionManager;
   }

   public CloseableHttpClient getHttpclient() {
      return httpclient;
   }


   @Override
   public void afterPropertiesSet() throws Exception {
      
      this.connectionManager = new PoolingHttpClientConnectionManager();
      this.connectionManager.setMaxTotal(this.maxTotal);
      this.connectionManager.setDefaultMaxPerRoute(this.defaultMaxPerRoute);
      
      this.httpclient = HttpClients.custom().setConnectionManager(this.connectionManager).build();

   }

}
‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍

The java service task - ThreadSaveFM3RestServiceTask

public class ThreadSaveFM3RestServiceTask implements JavaDelegate {

    protected final Logger LOGGER = LoggerFactory.getLogger(getClass());
    
    private Expression api_key;

    private Expression rest_url;
    
    private Expression json_string;

    private Expression http_method;
    
   @Override
   public void execute(DelegateExecution execution) throws Exception {

      String final_api_key = (String)api_key.getValue(execution);
      String final_rest_url = (String)rest_url.getValue(execution);
      String final_json_string = (String)json_string.getValue(execution);
      String final_method = (String)http_method.getValue(execution);

      ApplicationContext ctx = ApplicationContextProvider.getApplicationContext();
      
                ServiceProvider fm3RestServiceProvider = (FM3RestServiceProviderImpl) ctx.getBean("fm3RestServiceProvider");
      PooledHttpClient pooledHttpClient = (PooledHttpClient) ctx.getBean("pooledHttpClient");
      
      try {  
           
            // prepare the request
            HttpEntityEnclosingRequestBase request = null;
            
            if(final_method.equals("POST"))
            {
               request = new HttpPost(fm3RestServiceProvider.getServiceUrl() + final_rest_url);
            }
               
            if(final_method.equals("PUT"))
            {
               request = new HttpPut(fm3RestServiceProvider.getServiceUrl() + final_rest_url);
            }
            
            // prepare headers
            request.addHeader("Accept", "application/json");
            request.addHeader("Content-Type", "application/json; charset=utf-8");
            request.addHeader("Authorization", "ApiKey " + final_api_key);
                             
            // prepare the body       
            StringEntity entity = new StringEntity(final_json_string, "UTF-8");    
            request.setEntity(entity);
            
            CloseableHttpResponse response = pooledHttpClient.getHttpclient().execute(request);
            
            if (response.getStatusLine().getStatusCode() != 200) 
            {
               LOGGER.debug("FM3RestServiceTask request status: " + response.getStatusLine().getStatusCode());            
            }
               
            try {  

               HttpEntity responseEntity = response.getEntity();  
                   
            } finally {  

               LOGGER.debug("FM3RestServiceTask close response");
               response.close();  
               request.releaseConnection();
            }
         
         
      } catch (ClientProtocolException e) 
      {
         LOGGER.debug("FM3RestServiceTask ClientProtocolException: " + e.getMessage());
         
      } catch (IOException e) 
      {
         LOGGER.debug("FM3RestServiceTask IOException: " + e.getMessage());

      } finally 
      {
         LOGGER.debug("FM3RestServiceTask done");
      }
      
   }
}
‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍

The usage of the service task in a process definition

    <serviceTask id="servicetask1" name="FM3 Rest Service Task" activiti:async="true" activiti:class="com.sw.activiti.service.tasks.rest.ThreadSaveFM3RestServiceTask">
      <extensionElements>
        <activiti:field name="rest_url">
          <activiti:expression>Termine/setTerminStatus/${terminID}</activiti:expression>
        </activiti:field>
        <activiti:field name="api_key">
          <activiti:string><![CDATA[myapikey]]></activiti:string>
        </activiti:field>
        <activiti:field name="json_string">
          <activiti:string><![CDATA[{"terminStatus3":70,"terminStatus3Info":"Ressource ueberfaellig"}]]></activiti:string>
        </activiti:field>
        <activiti:field name="http_method">
          <activiti:string><![CDATA[POST]]></activiti:string>
        </activiti:field>
      </extensionElements>
    </serviceTask>
‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍

[martinfranz]

Update: I almost solved the problem.

I forgot to mention that I run actviti-rest and activiti-explorer next together in one tomcat instance. Both use the same mysql "activiti" schema.

I double checked the processEngineConfiguration in both application context files and realized that my activate-rest uses
jobExecutorActivate = false. So I set this to true in both context files.

What happens now:

1. Start 100 process instances of a simple process containing a timer intermediate event and my service task via REST API.
2. Login to activiti-exlporer as admin and can see 100 timer jobs firing in x minutes.
3. ACT_RU_JOB from mysql also shows me these 100 jobs.
4. Than the timer event occurs and i can see in ACT_RU_JOB that the jobs are done.
5. After 3-4 minutes all jobs are done and ACT_RU_JOB is empty.
6. Check the network traffic on the target API side and realized that all 100 calls where made.
7. Everything OK
8. Try to login into activate-explorer again. Freeze! Communication failure.

Nothing happens here. Activiti-explorer is out of reach.
But activit-rest works as aspected. (same time same machine) I can serve task or process definitions via rest api without any problem.

Restart activiti-explorer via tomcat manager does not work.

Shutdown tomcat. Killall java. Startup tomcat. Works! Activit-explorer is now running again.

JAVA_OPTS
-Xms1024m -Xmx2048m -XXermSize=32m -XX:MaxPermSize=512m -Xss2m -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -Djava.awt.headless=true

[jbarrez]

Hmmm you are not the first one reporting this problem. It seems there is a fundamental problem with the job executor + Activiti Explorer in the latest releases. But so far, we haven't found the cause yet.

Thanks for digging into this, much appreciated. So your analysis points in the direction of the threadpool on the Tomcat rather than the database connection pool?

[martinfranz]

Could it be a problem that both the activiti-explorer as well as the activiti-rest app running on one tomcat instance?

I mean - with this setup you have two running activiti engines pointing to the same database. Pooling Jobs from one table? Could this work without any problems?

In tomcat you can specify an executor with a max thread size.
<code>
<Executor name="tomcatThreadPool" namePrefix="catalina-exec-" maxThreads="150" minSpareThreads="4"/>
</code>

I made another test:

1. Stop activiti-explorer
2. Start 400 process instances via activiti-rest. (in a for loop, whit a delay of 1 second per iteration)
The process definition here is quit simple:
StartEvent > Timer Caching Event > Java Service Task > User Task > End Event

3. ACT_RU_JOB now shows 400 entries (400 timers - they all should fire at the same time)
4. Timer event fires and the number of entries in ACT_RU_JOB slowly begins to decrease.
5. After 12 Minutes all Jobs are done.
6. Startup activiti-explorer via tomcat manager.
7. Everything works as aspected.

The problem with this test for me is that the execution takes to long. It feels like all process instances run in sequence - not parallel.
I tried async true and false on the java service task. Same result.
Normally there should be at least 100 http client executions at the same time. But here there is just one by one. … But this could also be a problem within the connection pool ore with my java service task implementation themselves.

I make some more tests an keep you up to date.

Best regards from Nuremberg

[jbarrez]

Thanks, good and well documented explanation.

Just for validation: if you do this on Activiti 5.15, does it happen also? I have a hunch it might be due some 'logic' that was added there.

However, the async true/false that you tried worries me.

Could you hook up jvisualvm to the instance and see if it has indeed 100 threads at least blocked?

[martinfranz]

So first of all - sorry for mention that so late - i am currently using Version 5.15.1.

And the second thing: After a bit more digging into this problem I would say that activiti does his job well. My problem here was how I used the HttpClient and also the misunderstanding how Java Service Tasks are executed.


Just imagine the following scenario:

You have a Java Service Task creating and executing a http client directly in the execute method. That could not work. Even the apache docs for http client say that you should never use multiple instances of http clients.
> What happens here: The http client execution blocks the thread which executes the Java Service Task.
> You want recognize this as long as your Service Task comes after a User Task. In this case it is definitely not possible that multiple user execute their task absolutely synchron.
> You definitely will recognize this if your Service Task comes a the beginning of a process or after a timer event. In this case it could happens that the Service Task is executed a 100 times at the same time.


I ended up with the following setup:

1. Create a global connection pool for all your http client executions.
2. Limit the max client connections! (Less is more)
3. Create a global instance of a http client using the connection pool as well as other strategy implementations. (keep alive, connection reuse)
4. In your Java Service Task create the desired http method with headers etc as well as a new thread which executes your method with your single instance http client.

I have tested this scenario with 400 concurrent executions - means 400 process instances starting at the same time.
> Start Event > Timer Event > Java Service Task > User Task.
Results:
1. After the 400 timers fired activiti needed just 2 seconds the generate the 400 User Task. (No blocking anymore)
2. The connection pool with max 20 connections needed about 2 minutes for executing all requests.

I feel save now using the http client within a service task 🙂

There is just one more thing i have not tested jet:
I have started this post because my initial problem was the usage of acitvit rest and activiti explorer in one tomcat instance with on db. I don't know much about the core functionality of the activati engine but could it be possible that two job executors pulling jobs form the same table
won't work?

Best regards.

[jbarrez]

Ok, thanks for posting this all back. Very good information for many people!

Two job executors normally have no problems pulling from the same database (it was designed for that).