SimpleExecutionRunnable implements Runnable { @Override public void run() { String executionId = executionMessage.getMsgId(); String origThreadName = currentThread().getName(); currentThread().setName(origThreadName + "_" + executionId); Execution execution = null; try { execution = (executionMessage.getExecutionObject() != null) ? executionMessage.getExecutionObject() : converter.extractExecution(executionMessage.getPayload()); if (executionService.isSplitStep(execution)) { executeSplitStep(execution); } else { executeRegularStep(execution); } } catch (InterruptedException interruptedException) { if (workerManager.isFromCurrentThreadPool(currentThread().getName()) && isExecutionCancelled(execution)) { if (logger.isDebugEnabled()) { logger.debug("Execution is interrupted..."); } } else { logger.error("Execution thread is interrupted!!! Exiting...", interruptedException); } } catch (Exception ex) { logger.error("Error during execution!!!", ex); executionMessage.setStatus(ExecStatus.FAILED); executionMessage .incMsgSeqId(); try { if (executionMessage.getPayload() == null) { executionMessage.setPayload(converter.createPayload(execution)); } outBuffer.put(executionMessage); } catch (InterruptedException e) { logger.warn("Thread was interrupted! Exiting the execution... ", ex); } } finally { endExecutionCallback.endExecution(parseLong(executionId)); currentThread().setName(origThreadName); } } SimpleExecutionRunnable(ExecutionService executionService,
                                   OutboundBuffer outBuffer,
                                   InBuffer inBuffer,
                                   ExecutionMessageConverter converter,
                                   EndExecutionCallback endExecutionCallback,
                                   QueueStateIdGeneratorService queueStateIdGeneratorService,
                                   String workerUUID,
                                   WorkerConfigurationService workerConfigurationService,
                                   WorkerManager workerManager); ExecutionMessage getExecutionMessage(); void setExecutionMessage(ExecutionMessage executionMessage); @Override void run();  }
@Test public void testRun() throws Exception { Execution execution = new Execution(); when(converter.extractExecution(any(Payload.class))).thenReturn(execution); final List<ExecutionMessage> buffer = new ArrayList<>(); doAnswer(invocation -> { for (Object message : invocation.getArguments()) { buffer.add((ExecutionMessage) message); } return null; }).when(outBuffer).put(any(ExecutionMessage[].class)); when(workerManager.isFromCurrentThreadPool(anyString())).thenReturn(true); SimpleExecutionRunnable simpleExecutionRunnable = new SimpleExecutionRunnable(executionService, outBuffer, inBuffer, converter, endExecutionCallback, queueStateIdGenerator, "stam", workerConfigurationService, workerManager); ExecutionMessage executionMessage = new ExecutionMessage(); executionMessage.setMsgId(String.valueOf(100L)); simpleExecutionRunnable.setExecutionMessage(executionMessage); simpleExecutionRunnable.run(); verify(executionService, times(1)).execute(execution); Assert.assertFalse(buffer.isEmpty()); assertEquals(ExecStatus.FINISHED, buffer.get(0).getStatus()); assertEquals(ExecStatus.FINISHED, buffer.get(0).getStatus()); assertEquals(0, this.executionMessage.getMsgSeqId()); }