test_class.file: step-plans/step-plans-core/src/test/java/step/threadpool/ThreadPoolTest.java
test_class.identifier: ThreadPoolTest
test_case.body: @Test
  	public void testContextInterrupt() {
  		ExecutionContext context = newExecutionContext();
  		ThreadPool threadPool = new ThreadPool(context);
  		
  		List<String> itemList = Arrays.asList("item1", "item2", "item3");
  		
  		List<String> processedItems = new ArrayList<>();
  		
  		threadPool.consumeWork(itemList.iterator(), new WorkerItemConsumerFactory<String>() {
  			@Override
  			public Consumer<String> createWorkItemConsumer(WorkerController<String> control) {
  				return item -> {
  					processedItems.add(item);
  					if(item.equals("item2")) {
  						context.updateStatus(ExecutionStatus.ABORTING);
  					}
  				};
  			}
  		}, 1);
  		
  		Assert.assertEquals(Arrays.asList("item1", "item2"), processedItems);
  	}
test_case.class_method_signature: ThreadPoolTest.testContextInterrupt()
test_case.full_signature: @Test public void testContextInterrupt()
test_case.identifier: testContextInterrupt
test_case.invocations: newExecutionContext asList consumeWork iterator add equals updateStatus assertEquals asList
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testContextInterrupt()
focal_class.declarator: EXECUTION_THREADS_AUTO = "execution_threads_auto"
focal_class.modifier: private static final
focal_class.original_string: private static final String EXECUTION_THREADS_AUTO = "execution_threads_auto";
focal_class.type: String
focal_class.var_name: EXECUTION_THREADS_AUTO
focal_class.declarator: logger = LoggerFactory.getLogger(ThreadPool.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger logger = LoggerFactory.getLogger(ThreadPool.class);
focal_class.type: Logger
focal_class.var_name: logger
focal_class.declarator: executionContext
focal_class.modifier: private final
focal_class.original_string: private final ExecutionContext executionContext;
focal_class.type: ExecutionContext
focal_class.var_name: executionContext
focal_class.declarator: executorService = Executors.newCachedThreadPool()
focal_class.modifier: private final
focal_class.original_string: private final ExecutorService executorService = Executors.newCachedThreadPool();
focal_class.type: ExecutorService
focal_class.var_name: executorService
focal_class.declarator: batchContextStack = ThreadLocal.withInitial(()->new Stack<BatchContext>())
focal_class.modifier: protected
focal_class.original_string: protected ThreadLocal<Stack<BatchContext>> batchContextStack = ThreadLocal.withInitial(()->new Stack<BatchContext>());
focal_class.type: ThreadLocal<Stack<BatchContext>>
focal_class.var_name: batchContextStack
focal_class.file: step-plans/step-plans-core/src/main/java/step/threadpool/ThreadPool.java
focal_class.identifier: ThreadPool
focal_class.class_method_signature: ThreadPool.ThreadPool(ExecutionContext context)
focal_class.full_signature: public  ThreadPool(ExecutionContext context)
focal_class.identifier: ThreadPool
focal_class.modifiers: public
focal_class.parameters: (ExecutionContext context)
focal_class.signature:  ThreadPool(ExecutionContext context)
focal_class.class_method_signature: ThreadPool.consumeWork(Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads)
focal_class.full_signature: public void consumeWork(Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads)
focal_class.identifier: consumeWork
focal_class.modifiers: public
focal_class.parameters: (Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads)
focal_class.return: void
focal_class.signature: void consumeWork(Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads)
focal_class.class_method_signature: ThreadPool.isReentrantThread()
focal_class.full_signature: protected boolean isReentrantThread()
focal_class.identifier: isReentrantThread
focal_class.modifiers: protected
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean isReentrantThread()
focal_class.class_method_signature: ThreadPool.getAutoNumberOfThreads()
focal_class.full_signature: protected Integer getAutoNumberOfThreads()
focal_class.identifier: getAutoNumberOfThreads
focal_class.modifiers: protected
focal_class.parameters: ()
focal_class.return: Integer
focal_class.signature: Integer getAutoNumberOfThreads()
focal_class.class_method_signature: ThreadPool.createWorkerAndRun(BatchContext batchContext, Consumer<WORK_ITEM> workItemConsumer, Iterator<WORK_ITEM> workItemIterator, int workerId)
focal_class.full_signature: private void createWorkerAndRun(BatchContext batchContext, Consumer<WORK_ITEM> workItemConsumer, Iterator<WORK_ITEM> workItemIterator, int workerId)
focal_class.identifier: createWorkerAndRun
focal_class.modifiers: private
focal_class.parameters: (BatchContext batchContext, Consumer<WORK_ITEM> workItemConsumer, Iterator<WORK_ITEM> workItemIterator, int workerId)
focal_class.return: void
focal_class.signature: void createWorkerAndRun(BatchContext batchContext, Consumer<WORK_ITEM> workItemConsumer, Iterator<WORK_ITEM> workItemIterator, int workerId)
focal_class.class_method_signature: ThreadPool.pushBatchContextToStack(final BatchContext batchContext)
focal_class.full_signature: protected Stack<BatchContext> pushBatchContextToStack(final BatchContext batchContext)
focal_class.identifier: pushBatchContextToStack
focal_class.modifiers: protected
focal_class.parameters: (final BatchContext batchContext)
focal_class.return: Stack<BatchContext>
focal_class.signature: Stack<BatchContext> pushBatchContextToStack(final BatchContext batchContext)
focal_method.body: public <WORK_ITEM> void consumeWork(Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads) {
  		// Wrapping the iterator to avoid concurrency issues as iterators aren't ThreadSafe 
  		Iterator<WORK_ITEM> threadSafeIterator = new Iterator<WORK_ITEM>() {
  			@Override
  			public boolean hasNext() {
  				throw new RuntimeException("This method shouldn't be called");
  			}
  
  			@Override
  			public WORK_ITEM next() {
  				synchronized (this) {
  					return workItemIterator.next();
  				}
  			}
  		};
  		
  		Integer autoNumberOfThreads = getAutoNumberOfThreads();
  		if (autoNumberOfThreads != null) {
  			if(!isReentrantThread()) {
  				// Forcing the number of threads to the required autoNumberOfThreads for the 
  				// first Artefact using the ThreadPool (Level = 1)
  				numberOfThreads = autoNumberOfThreads;
  			} else {
  				// Avoid parallelism for the artefacts that are children of an artefact 
  				// already using the ThreadPool (Level > 1)
  				numberOfThreads = 1;
  			}
  		}
  		
  		final BatchContext batchContext = new BatchContext(executionContext, numberOfThreads>1);
  		
  		WorkerController<WORK_ITEM> workerController = new WorkerController<>(batchContext);
  		Consumer<WORK_ITEM> workItemConsumer = workItemConsumerFactory.createWorkItemConsumer(workerController);
  		
  		if(numberOfThreads == 1) {
  			// No parallelism, run the worker in the current thread
  			createWorkerAndRun(batchContext, workItemConsumer, threadSafeIterator, 0);
  		} else {
  			ReportNode currentReportNode = executionContext.getCurrentReportNode();
  			List<Future<?>> futures = new ArrayList<>();
  			long parentThreadId = Thread.currentThread().getId();
  			// Create one worker for each "thread"
  			for (int i = 0; i < numberOfThreads; i++) {
  				int workerId = i;
  				futures.add(executorService.submit(() -> {
  					executionContext.associateThread(parentThreadId, currentReportNode);
  					createWorkerAndRun(batchContext, workItemConsumer, threadSafeIterator, workerId);
  				}));
  			}
  			
  			// Wait for the workers to complete
  			for (Future<?> future : futures) {
  				try {
  					future.get();
  				} catch (InterruptedException | ExecutionException e) {
  					logger.error("Error while waiting for worker execution to terminate. Execution ID: "+executionContext.getExecutionId(), e);
  				}
  			}
  		}
  	}
focal_method.class_method_signature: ThreadPool.consumeWork(Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads)
focal_method.full_signature: public void consumeWork(Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads)
focal_method.identifier: consumeWork
focal_method.invocations: next getAutoNumberOfThreads isReentrantThread createWorkItemConsumer createWorkerAndRun getCurrentReportNode getId currentThread add submit associateThread createWorkerAndRun get error getExecutionId
focal_method.modifiers: public
focal_method.parameters: (Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads)
focal_method.return: void
focal_method.signature: void consumeWork(Iterator<WORK_ITEM> workItemIterator,
  			WorkerItemConsumerFactory<WORK_ITEM> workItemConsumerFactory, int numberOfThreads)