<commit_before>/* Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.activiti.engine.test.jobexecutor;

import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.activiti.engine.ActivitiException;
import org.activiti.engine.ProcessEngine;
import org.activiti.engine.impl.asyncexecutor.AsyncExecutor;
import org.activiti.engine.impl.asyncexecutor.DefaultAsyncJobExecutor;
import org.activiti.engine.impl.cfg.ProcessEngineConfigurationImpl;
import org.activiti.engine.impl.cfg.StandaloneInMemProcessEngineConfiguration;
import org.activiti.engine.impl.persistence.entity.JobEntity;
import org.activiti.engine.impl.test.JobTestHelper;
import org.junit.Assert;
import org.junit.Test;

/**
 * Tests specifically for the {@link AsyncExecutor}.
 * 
 * Note that all tests with jobs already use this async executor, so this
 * is really all the 'edgy' cases here. 
 * 
 * @author jbarrez
 */
public class AsyncExecutorTest {
	
	@Test
	public void testRegularAsyncExecution() {
		
		// Deploy
		ProcessEngine processEngine = createProcessEngine(true);
		setClockToCurrentTime(processEngine);
		deploy(processEngine, "AsyncExecutorTest.testRegularAsyncExecution.bpmn20.xml");
		
		// Start process instance. Wait for all jobs to be done
		processEngine.getRuntimeService().startProcessInstanceByKey("asyncExecutor");
		
		// Move clock 3 minutes. Nothing should happen
		addSecondsToCurrentTime(processEngine, 180L);
		try {
			waitForAllJobsBeingExecuted(processEngine, 500L);
			Assert.fail();
		} catch (ActivitiException e) {
			// Expected
		}
		Assert.assertEquals(1, processEngine.getTaskService().createTaskQuery().taskName("The Task").count());
		Assert.assertEquals(0, processEngine.getTaskService().createTaskQuery().taskName("Task after timer").count());
		Assert.assertEquals(1, processEngine.getManagementService().createJobQuery().count());

		// Move clock 3 minutes and 1 second. Triggers the timer
		addSecondsToCurrentTime(processEngine, 181); 
		waitForAllJobsBeingExecuted(processEngine);

		// Verify if all is as expected
		Assert.assertEquals(0, processEngine.getTaskService().createTaskQuery().taskName("The Task").count());
		Assert.assertEquals(1, processEngine.getTaskService().createTaskQuery().taskName("Task after timer").count());
		Assert.assertEquals(0, processEngine.getManagementService().createJobQuery().count());
		
		Assert.assertEquals(1, getAsyncExecutorJobCount(processEngine));
		
		// Clean up
		cleanup(processEngine);
		
	}
	
	@Test
	public void testAsyncExecutorDisabledOnOneEngine() {
		
		// Deploy on one engine, where the async executor is disabled
		ProcessEngine firstProcessEngine = createProcessEngine(false);
		Date now = setClockToCurrentTime(firstProcessEngine);
		deploy(firstProcessEngine, "AsyncExecutorTest.testRegularAsyncExecution.bpmn20.xml");
		
		// Start process instance on first engine
		firstProcessEngine.getRuntimeService().startProcessInstanceByKey("asyncExecutor");
		
		//	Move clock 5 minutes and 1 second. Triggers the timer normally, but not now since async execution is disabled
		addSecondsToCurrentTime(firstProcessEngine, 301); // 301 = 5m01s
		Assert.assertEquals(1, firstProcessEngine.getTaskService().createTaskQuery().taskName("The Task").count());
		Assert.assertEquals(0, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after timer").count());
		Assert.assertEquals(1, firstProcessEngine.getManagementService().createJobQuery().count());
		
		// Create second engine, with async executor enabled. Same time as the first engine to start, then add 301 seconds
		ProcessEngine secondProcessEngine = createProcessEngine(true, now);
		addSecondsToCurrentTime(secondProcessEngine, 301); // 301 = 5m01s
		waitForAllJobsBeingExecuted(secondProcessEngine);

		// Verify if all is as expected
		Assert.assertEquals(0, firstProcessEngine.getTaskService().createTaskQuery().taskName("The Task").count());
		Assert.assertEquals(1, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after timer").count());
		Assert.assertEquals(0, firstProcessEngine.getManagementService().createJobQuery().count());
		
		Assert.assertEquals(0, getAsyncExecutorJobCount(firstProcessEngine));
		Assert.assertEquals(1, getAsyncExecutorJobCount(secondProcessEngine));
		
		// Clean up
		cleanup(firstProcessEngine);
		cleanup(secondProcessEngine);
		
	}
	
	@Test
	public void testAsyncScriptExecution() {

		// Deploy
		ProcessEngine processEngine = createProcessEngine(true);
		setClockToCurrentTime(processEngine);
		deploy(processEngine,"AsyncExecutorTest.testAsyncScriptExecution.bpmn20.xml");

		// Start process instance. Wait for all jobs to be done
		processEngine.getRuntimeService().startProcessInstanceByKey("asyncScript");
		waitForAllJobsBeingExecuted(processEngine);

		// Verify if all is as expected
		Assert.assertEquals(1, processEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
		Assert.assertEquals(0, processEngine.getManagementService().createJobQuery().count());

		Assert.assertEquals(1, getAsyncExecutorJobCount(processEngine));
		
		// Clean up
		cleanup(processEngine);
		
	}
	
	@Test
	public void testAsyncScriptExecutionOnTwoEngines() {

		// Deploy
		ProcessEngine firstProcessEngine = createProcessEngine(false);
		Date now = setClockToCurrentTime(firstProcessEngine);
		deploy(firstProcessEngine,"AsyncExecutorTest.testAsyncScriptExecution.bpmn20.xml");

		// Start process instance. Nothing should happen
		firstProcessEngine.getRuntimeService().startProcessInstanceByKey("asyncScript");
		Assert.assertEquals(0, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
		Assert.assertEquals(1, firstProcessEngine.getManagementService().createJobQuery().count());
		
		// Start second engine, with async executor enabled
		ProcessEngine secondProcessEngine = createProcessEngine(true, now); // Same timestamp as first engine
		Assert.assertEquals(0, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
		Assert.assertEquals(1, firstProcessEngine.getManagementService().createJobQuery().count());
		
		// Move the clock 1 second. Should be executed now by second engine
		addSecondsToCurrentTime(secondProcessEngine, 1);
		waitForAllJobsBeingExecuted(secondProcessEngine, 10000L);

		// Verify if all is as expected
		Assert.assertEquals(1, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
		Assert.assertEquals(0, firstProcessEngine.getManagementService().createJobQuery().count());

		Assert.assertEquals(0, getAsyncExecutorJobCount(firstProcessEngine));
		Assert.assertEquals(1, getAsyncExecutorJobCount(secondProcessEngine));
		
		// Clean up
		cleanup(firstProcessEngine);
		
	}
	
	@Test
	public void testAsyncFailingScript() {
		
		// Deploy
		final ProcessEngine processEngine = createProcessEngine(true);
		processEngine.getProcessEngineConfiguration().getClock().reset();
		deploy(processEngine,"AsyncExecutorTest.testAsyncFailingScript.bpmn20.xml");

		// Start process instance. Wait for all jobs to be done.
		processEngine.getRuntimeService().startProcessInstanceByKey("asyncScript");
		
		// There is a back off mechanism for the retry, so need a bit of time
		// But to be sure, we make the wait time small 
		processEngine.getProcessEngineConfiguration().setAsyncFailedJobWaitTime(1);
		
		JobTestHelper.waitForJobExecutorOnCondition(processEngine.getProcessEngineConfiguration(), 10000L, 2000L, new Callable<Boolean>() {
			@Override
			public Boolean call() throws Exception {
				return processEngine.getManagementService().createJobQuery().withRetriesLeft().count() == 0;
			}
		});
		

		// Verify if all is as expected
		Assert.assertEquals(0, processEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
		Assert.assertEquals(1, processEngine.getManagementService().createJobQuery().count());
		Assert.assertEquals(0, processEngine.getManagementService().createJobQuery().withRetriesLeft().count());
		Assert.assertEquals(1, processEngine.getManagementService().createJobQuery().noRetriesLeft().count());
		Assert.assertEquals(1, processEngine.getManagementService().createJobQuery().withException().count());

		Assert.assertEquals(3, getAsyncExecutorJobCount(processEngine));
		
		// Clean up
		cleanup(processEngine);
		
	}
	
	
	
	// Helpers
	
	private ProcessEngine createProcessEngine(boolean enableAsyncExecutor) {
		return createProcessEngine(enableAsyncExecutor, null);
	}
	
	private ProcessEngine createProcessEngine(boolean enableAsyncExecutor, Date time) {
		ProcessEngineConfigurationImpl processEngineConfiguration = new StandaloneInMemProcessEngineConfiguration();
		processEngineConfiguration.setJdbcUrl("jdbc:h2:mem:activiti-AsyncExecutorTest;DB_CLOSE_DELAY=1000");
		processEngineConfiguration.setDatabaseSchemaUpdate("true");
		processEngineConfiguration.setJobExecutorActivate(false); // No need for that old job executor
		
		if (enableAsyncExecutor) {
			processEngineConfiguration.setAsyncExecutorEnabled(true);
			processEngineConfiguration.setAsyncExecutorActivate(true);<commit_msg>Fix for failing AsyncExecutorTest (part2 - always cleaning up)<commit_after>/* Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.activiti.engine.test.jobexecutor;

import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.activiti.engine.ActivitiException;
import org.activiti.engine.ProcessEngine;
import org.activiti.engine.impl.asyncexecutor.AsyncExecutor;
import org.activiti.engine.impl.asyncexecutor.DefaultAsyncJobExecutor;
import org.activiti.engine.impl.cfg.ProcessEngineConfigurationImpl;
import org.activiti.engine.impl.cfg.StandaloneInMemProcessEngineConfiguration;
import org.activiti.engine.impl.persistence.entity.JobEntity;
import org.activiti.engine.impl.test.JobTestHelper;
import org.junit.Assert;
import org.junit.Test;

/**
 * Tests specifically for the {@link AsyncExecutor}.
 * 
 * Note that all tests with jobs already use this async executor, so this
 * is really all the 'edgy' cases here. 
 * 
 * @author jbarrez
 */
public class AsyncExecutorTest {
	
	@Test
	public void testRegularAsyncExecution() {
		
		ProcessEngine processEngine = null;
		
		try {
			// Deploy
			processEngine = createProcessEngine(true);
			setClockToCurrentTime(processEngine);
			deploy(processEngine, "AsyncExecutorTest.testRegularAsyncExecution.bpmn20.xml");
			
			// Start process instance. Wait for all jobs to be done
			processEngine.getRuntimeService().startProcessInstanceByKey("asyncExecutor");
			
			// Move clock 3 minutes. Nothing should happen
			addSecondsToCurrentTime(processEngine, 180L);
			try {
				waitForAllJobsBeingExecuted(processEngine, 500L);
				Assert.fail();
			} catch (ActivitiException e) {
				// Expected
			}
			Assert.assertEquals(1, processEngine.getTaskService().createTaskQuery().taskName("The Task").count());
			Assert.assertEquals(0, processEngine.getTaskService().createTaskQuery().taskName("Task after timer").count());
			Assert.assertEquals(1, processEngine.getManagementService().createJobQuery().count());
	
			// Move clock 3 minutes and 1 second. Triggers the timer
			addSecondsToCurrentTime(processEngine, 181); 
			waitForAllJobsBeingExecuted(processEngine);
	
			// Verify if all is as expected
			Assert.assertEquals(0, processEngine.getTaskService().createTaskQuery().taskName("The Task").count());
			Assert.assertEquals(1, processEngine.getTaskService().createTaskQuery().taskName("Task after timer").count());
			Assert.assertEquals(0, processEngine.getManagementService().createJobQuery().count());
			
			Assert.assertEquals(1, getAsyncExecutorJobCount(processEngine));
		} finally {
		
			// Clean up
			if (processEngine != null) {
				cleanup(processEngine);
			}
			
		}
		
	}
	
	@Test
	public void testAsyncExecutorDisabledOnOneEngine() {
		
		ProcessEngine firstProcessEngine = null;
		ProcessEngine secondProcessEngine = null;
		
		try {
		
			// Deploy on one engine, where the async executor is disabled
			firstProcessEngine = createProcessEngine(false);
			Date now = setClockToCurrentTime(firstProcessEngine);
			deploy(firstProcessEngine, "AsyncExecutorTest.testRegularAsyncExecution.bpmn20.xml");
			
			// Start process instance on first engine
			firstProcessEngine.getRuntimeService().startProcessInstanceByKey("asyncExecutor");
			
			//	Move clock 5 minutes and 1 second. Triggers the timer normally, but not now since async execution is disabled
			addSecondsToCurrentTime(firstProcessEngine, 301); // 301 = 5m01s
			Assert.assertEquals(1, firstProcessEngine.getTaskService().createTaskQuery().taskName("The Task").count());
			Assert.assertEquals(0, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after timer").count());
			Assert.assertEquals(1, firstProcessEngine.getManagementService().createJobQuery().count());
			
			// Create second engine, with async executor enabled. Same time as the first engine to start, then add 301 seconds
			secondProcessEngine = createProcessEngine(true, now);
			addSecondsToCurrentTime(secondProcessEngine, 301); // 301 = 5m01s
			waitForAllJobsBeingExecuted(secondProcessEngine);
	
			// Verify if all is as expected
			Assert.assertEquals(0, firstProcessEngine.getTaskService().createTaskQuery().taskName("The Task").count());
			Assert.assertEquals(1, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after timer").count());
			Assert.assertEquals(0, firstProcessEngine.getManagementService().createJobQuery().count());
			
			Assert.assertEquals(0, getAsyncExecutorJobCount(firstProcessEngine));
			Assert.assertEquals(1, getAsyncExecutorJobCount(secondProcessEngine));
			
		} finally {
			
			// Clean up
			cleanup(firstProcessEngine);
			cleanup(secondProcessEngine);
			
		}
		
	}
	
	@Test
	public void testAsyncScriptExecution() {
		
		ProcessEngine processEngine = null;
		
		try {

			// Deploy
			processEngine = createProcessEngine(true);
			setClockToCurrentTime(processEngine);
			deploy(processEngine,"AsyncExecutorTest.testAsyncScriptExecution.bpmn20.xml");
	
			// Start process instance. Wait for all jobs to be done
			processEngine.getRuntimeService().startProcessInstanceByKey("asyncScript");
			waitForAllJobsBeingExecuted(processEngine);
	
			// Verify if all is as expected
			Assert.assertEquals(1, processEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
			Assert.assertEquals(0, processEngine.getManagementService().createJobQuery().count());
	
			Assert.assertEquals(1, getAsyncExecutorJobCount(processEngine));
			
		} finally {
		
			// Clean up
			cleanup(processEngine);
			
		}
		
	}
	
	@Test
	public void testAsyncScriptExecutionOnTwoEngines() {
		
		ProcessEngine firstProcessEngine = null;
		ProcessEngine secondProcessEngine = null;
		
		try {

			// Deploy
			firstProcessEngine = createProcessEngine(false);
			Date now = setClockToCurrentTime(firstProcessEngine);
			deploy(firstProcessEngine,"AsyncExecutorTest.testAsyncScriptExecution.bpmn20.xml");
	
			// Start process instance. Nothing should happen
			firstProcessEngine.getRuntimeService().startProcessInstanceByKey("asyncScript");
			Assert.assertEquals(0, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
			Assert.assertEquals(1, firstProcessEngine.getManagementService().createJobQuery().count());
			
			// Start second engine, with async executor enabled
			secondProcessEngine = createProcessEngine(true, now); // Same timestamp as first engine
			Assert.assertEquals(0, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
			Assert.assertEquals(1, firstProcessEngine.getManagementService().createJobQuery().count());
			
			// Move the clock 1 second. Should be executed now by second engine
			addSecondsToCurrentTime(secondProcessEngine, 1);
			waitForAllJobsBeingExecuted(secondProcessEngine, 10000L);
	
			// Verify if all is as expected
			Assert.assertEquals(1, firstProcessEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
			Assert.assertEquals(0, firstProcessEngine.getManagementService().createJobQuery().count());
	
			Assert.assertEquals(0, getAsyncExecutorJobCount(firstProcessEngine));
			Assert.assertEquals(1, getAsyncExecutorJobCount(secondProcessEngine));
			
		} finally {
			
			// Clean up
			cleanup(firstProcessEngine);
			cleanup(secondProcessEngine);
			
		}
		
		
	}
	
	@Test
	public void testAsyncFailingScript() {
		
		ProcessEngine processEngine = null;
		
		try {
		
			// Deploy
			processEngine = createProcessEngine(true);
			processEngine.getProcessEngineConfiguration().getClock().reset();
			deploy(processEngine,"AsyncExecutorTest.testAsyncFailingScript.bpmn20.xml");
	
			// Start process instance. Wait for all jobs to be done.
			processEngine.getRuntimeService().startProcessInstanceByKey("asyncScript");
			
			// There is a back off mechanism for the retry, so need a bit of time
			// But to be sure, we make the wait time small 
			processEngine.getProcessEngineConfiguration().setAsyncFailedJobWaitTime(1);
			
			final ProcessEngine processEngineCopy = processEngine;
			JobTestHelper.waitForJobExecutorOnCondition(processEngine.getProcessEngineConfiguration(), 10000L, 2000L, new Callable<Boolean>() {
				@Override
				public Boolean call() throws Exception {
					return processEngineCopy.getManagementService().createJobQuery().withRetriesLeft().count() == 0;
				}
			});
			
	
			// Verify if all is as expected
			Assert.assertEquals(0, processEngine.getTaskService().createTaskQuery().taskName("Task after script").count());
			Assert.assertEquals(1, processEngine.getManagementService().createJobQuery().count());
			Assert.assertEquals(0, processEngine.getManagementService().createJobQuery().withRetriesLeft().count());
			Assert.assertEquals(1, processEngine.getManagementService().createJobQuery().noRetriesLeft().count());
			Assert.assertEquals(1, processEngine.getManagementService().createJobQuery().withException().count());
	
			Assert.assertEquals(3, getAsyncExecutorJobCount(processEngine));
			
		} finally {
			
			// Clean up
			cleanup(processEngine);
			
		}
		
	}
	
	
	
	// Helpers
	
	private ProcessEngine createProcessEngine(boolean enableAsyncExecutor) {
		return createProcessEngine(enableAsyncExecutor, null);
	}
	
	private ProcessEngine createProcessEngine(boolean enableAsyncExecutor, Date time) {
		ProcessEngineConfigurationImpl processEngineConfiguration = new StandaloneInMemProcessEngineConfiguration();
		processEngineConfiguration.setJdbcUrl("jdbc:h2:mem:activiti-AsyncExecutorTest;DB_CLOSE_DELAY=1000");
		processEngineConfiguration.setDatabaseSchemaUpdate("true");
		processEngineConfiguration.setJobExecutorActivate(false); // No need for that old job executor
		
		if (enableAsyncExecutor) {
			processEngineConfiguration.setAsyncExecutorEnabled(true);
			processEngineConfiguration.setAsyncExecutorActivate(true);
{'fraction_non_alphanumeric': 0.10341943419434195, 'fraction_numerical': 0.01028290282902829, 'mean_word_length': 5.502239283429303, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 10}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1121717', 'n_tokens_mistral': 6030, 'n_tokens_neox': 5226, 'n_words': 1187}