Activiti6&7 : Async Job Executor only executes job after lock expiration

[wuxianInfi]

In activiti6.0.0&7, the async service task does not start immediately. It can only be executed after the lock expiration.

Here is the log that show the problem:
It looks like the async service task is submitted to async executor without the database transaction in the main thread committed. So the job cannot be selected from database.
This is the same problem in stackoverflow but no accepted answer:
https://stackoverflow.com/questions/47535257/activiti-6-sometime-waits-5-minutes-before-starting-async-process-instance

*2019-03-27T16:20:31,473 [] [] [MyExecutor-1] DEBUG druid.sql.Connection - {conn-10008} setAutoCommit true
2019-03-27T16:20:31,473 [] [] [MyExecutor-2] DEBUG druid.sql.ResultSet - {conn-10009, pstmt-20038, rs-50026} closed
2019-03-27T16:20:31,473 [] [] [MyExecutor-2] DEBUG druid.sql.Statement - {conn-10009, pstmt-20038} closed
2019-03-27T16:20:31,473 [] [] [MyExecutor-2] DEBUG org.activiti.engine.impl.cmd.ExecuteAsyncJobCmd - Job does not exist anymore and will not be executed. It has most likely been deleted as part of another concurrent part of the process instance.
2019-03-27T16:20:31,473 [] [] [MyExecutor-2] DEBUG org.activiti.engine.impl.db.DbSqlSession - Flushing dbSqlSession
2019-03-27T16:20:31,473 [] [] [MyExecutor-2] DEBUG org.activiti.engine.impl.db.DbSqlSession - flush summary: 0 insert, 0 update, 0 delete.
2019-03-27T16:20:31,473 [] [] [MyExecutor-2] DEBUG org.activiti.engine.impl.db.DbSqlSession - now executing flush...
*

[rvaldeolmillos]

Having the same problem, using Activiti 6 and DefaultAsyncJobExecutor.
Some asynchronous continuations get delayed 5 minutes until the lock expires for no apparent reason.

[salaboy]

Ok, so this is not a problem in Activiti 7, because by default we are not using the AsyncJobExecutor.
It will be great if we can create a reproducer that we can use to investigate and solve the problem.

[jesperdj]

I have a problem that sounds the same as what's described above. I am using Activiti 6.0.0. with activiti-spring.

I took the time to dive into the source code and logging to find out exactly what is happening, and I have discovered what I believe is a bug - I'll try to explain it in detail below.

Activiti 6 has a pipeline of command interceptors to execute commands. This pipeline is built in ProcessEngineCongfigurationImpl.initCommandInterceptors() / getDefaultCommandInterceptors(). Normally, it will look like this if you are using activiti-spring (which means you are using SpringProcessEngineConfiguration):

LogInterceptor -> SpringTransactionInterceptor -> CommandContextInterceptor -> TransactionContextInterceptor (using SpringTransactionContextFactory) -> CommandInvoker

Note that SpringTransactionInterceptor is the thing that does the actual Spring transaction wrapping (by executing the rest of the pipeline inside a TransactionTemplate). And note, important for this bug as will become clear later, is that the SpringTransactionInterceptor is before the CommandContextInterceptor in the pipeline, so the SpringTransactionInterceptor wraps the CommandContextInterceptor.

The problem happens when an async job causes another async job to be scheduled. In that case, the method DefaultJobManager.scheduleAsyncJob() is called. This method inserts the new job entity and then calls triggerExecutorIfNeeded(). That method calls hintAsyncExecutor() which adds a close listener to the CommandContext; this close listener is an instance of AsyncJobAddedNotification.

Note that at this point, the new job entity is inserted into the database, but the transaction that contains the insert is still open. So the new job entity is not yet visible outside of the transaction!

When the current job finishes, the command executor pipeline winds back. The CommandInvoker passes control back to the CommandContextInterceptor, which passes back to the SpringTransactionInterceptor and finally the LogInterceptor.

The AsyncJobAddedNotification is called on the way back by the CommandContextInterceptor, when the CommandContext is closed. The method AsyncJobAddedNotification.closed() tells the async executor to schedule the new job that was added, for execution.

Here is where it goes wrong: It happens sometimes that one of the threads in the threadpool of the async executor almost immediately starts executing the new job. This can happen before the SpringTransactionInterceptor has committed the transaction of the original job. What happens then is that when executing the new job, the thread does not find the new job in the database. It then quits and the new job is left locked in the database.

After 5 minutes, the cleanup thread that unlocks stale jobs unlocks the job, and then next time the job acquisition thread comes along, it is executed.

For evidence, see the attached log file. You can see the following here:

• activiti-async-job-executor-thread-2 executes job 166
• this causes a new job to be created, job 170 (line 80)
• line 109: activiti-async-job-executor-thread-1 starts executing job 170 while thread-2 has not yet committed its transaction
• line 116 / 133: thread-1 does not find the job in the database
• the transaction of thread-2 is not committed until the end (line 175)

As you see, thread-1 picks up job 170 before thread-2's transaction has committed, so thread-1 does not find the job in the database.

[jesperdj]

Logfile, see comments above: activiti.log

[jesperdj]

Possible workarounds and solutions:

1. set transaction isolation level to dirty-read, so that the thread that executes the new job can see the job in the database even though the other thread has not yet committed its transaction => but using dirty-read is almost always a bad idea

2. do not use AsyncJobAddedNotification; just insert the new job in the database and let the job acquire thread find it next time it looks for jobs to execute

3. somehow make sure that the transaction is committed before AsyncJobAddedNotification is executed.

[oswulf75]

Possible workarounds and solutions:

1. set transaction isolation level to dirty-read, so that the thread that executes the new job can see the job in the database even though the other thread has not yet committed its transaction => but using dirty-read is almost always a bad idea

2. do not use AsyncJobAddedNotification; just insert the new job in the database and let the job acquire thread find it next time it looks for jobs to execute

3. somehow make sure that the transaction is committed before AsyncJobAddedNotification is executed.

I encountered the same problem in my application. In my deployments, i have my own application.ear and activiti-app.war running on the same wildfly instance. I have 2 questions:
1) How do i disable the async-executor in activiti-app as i am just using it as a modelling and deployment tool for my process definitions?
2) Can you elaborate a bit more on point 2 and 3 on the AsyncJobAddedNotification. Can we control this?

[jesperdj]

Hello @oswulf75

1. I don't think you can disable the async executor in Activiti 6, since in Activiti 6 it is the only executor implementation that is provided.
2. My suggestions were possible ways to fix it in the Activiti source code; these are not things you can easily do in your own application.

I have my own workaround which I came up with after studying the Activiti source code, but this workaround is complicated (too complicated to quickly explain here) and involves overriding a number of Activiti classes with custom versions.

[ngocthat012]

I also had the same problem, but when I debug it, I came to the newly created step and stood there waiting until 15p after it ran again and after 5p, there was another thread taken up and processed. I have seen the source but there is no solution.

Log:
2019-08-12 12:58:05,248 [pool-1-thread-3] DEBUG org.springframework.jdbc.datasource.DataSourceTransactionManager - Creating new transaction with name [null]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
2019-08-12 13:13:34,213 [pool-1-thread-3] DEBUG org.activiti.engine.impl.persistence.entity.JobEntity.selectJob - ==> Preparing: select * from ACT_RU_JOB where ID_ = ?
2019-08-12 13:13:34,213 [pool-1-thread-3] DEBUG org.activiti.engine.impl.persistence.entity.JobEntity.selectJob - ==> Parameters: 201378(String)
2019-08-12 13:13:34,215 [pool-1-thread-3] DEBUG org.activiti.engine.impl.persistence.entity.JobEntity.selectJob - <== Total: 0
2019-08-12 13:13:34,215 [pool-1-thread-3] INFO org.activiti.engine.impl.interceptor.CommandContext - Error while closing command context
org.activiti.engine.JobNotFoundException: No job found with id '201378'.

[ngocthat012]

Hi jesperdj,

Please provide the source code you fixed so that I can learn the source code?

[jesperdj]

Hi @ngocthat012 , here is my source code with a workaround.

The main idea here is that we add another command interceptor in the chain, which is outside the Spring transaction interceptor. The custom job manager, in its hintAsyncExecutor() method, registers a CustomAsyncJobAddedNotification to the context of the custom command interceptor. This will be called outside of the Spring transaction, by the CustomPreCommandInterceptor.
By doing it that way, the new job to be executed is passed to the async executor outside of the Spring transaction, so that the bug that I described in detail above doesn't happen anymore.

The custom job executor and custom pre-command interceptor have to be set on the process engine configuration, see class ActivitiConfiguration.

I hope it's useful.

workaround.tar.gz

[ngocthat012]

Thank @jesperdj

[jjava]

The method that start async job can not be transactional. You can change the spring transaction config ,for example ,remove the annotation @Transactional from the method.

[MephistoMMM]

The method that start async job can not be transactional. You can change the spring transaction config ,for example ,remove the annotation @transactional from the method.

But starting processes and completing tasks are almost transactional, for rollback while failed.

[Klapsa2503]

Hello, i came up with two workarounds

1 Involves adding following post commad interceptor:

springProcessEngineConfiguration.setCustomPostCommandInterceptors(Lists.newArrayList(new AbstractCommandInterceptor() {

                @Override
                public <T> T execute(CommandConfig commandConfig, Command<T> command) {
                    try {
                        return next.execute(commandConfig, command);
                    } finally {

                        List<CommandContextCloseListener> listeners = new ArrayList<>();
                        CommandContext customCommandContext = Context.getCommandContext();
                        if (customCommandContext.getCloseListeners() != null) {
                            for (int i = customCommandContext.getCloseListeners().size() - 1; i > 0; i--) {
                                CommandContextCloseListener listener = customCommandContext.getCloseListeners().get(i);
                                if (listener instanceof AsyncJobAddedNotification) {
                                    listeners.add(listener);
                                    customCommandContext.getCloseListeners().remove(i);
                                }
                            }
                        }

                        TransactionContext transactionContext = Context.getTransactionContext();
                        transactionContext.addTransactionListener(
                                TransactionState.COMMITTED,
                                commandContext -> listeners.forEach(listener -> listener.closed(commandContext))
                        );
                    }
                }
}));

This command interceptor converts all AsyncJobAddedNotification into TransactionListeners resulting in AsyncJobAddedNotification being executed after transaction commits.

2 Involves adding following job manager

springProcessEngineConfiguration.setJobManager(new DefaultJobManager(springProcessEngineConfiguration) {
                @Override
                protected void hintAsyncExecutor(JobEntity job) {
                   log.info("Ommiting AsyncJobAddedNotification for job {}", job.getId());
                }

                @Override
                public JobEntity createAsyncJob(ExecutionEntity execution, boolean exclusive) {
                    return internalCreateAsyncJob(execution, exclusive);
                }
});

This job manager disables use of AsyncJobAddedNotification and creates new async jobs without locks (they will be executed by AcquireAsyncJobsDueRunnable)

Please verify if those workaround are sufficient in your cases and if there are some risks / downsides

[erdemedeiros]

This should get fixed by https://github.com/Activiti/Activiti/pull/3367.