@Test public void handlePotentiallyStuckInterruptsThreadWhenListenerReturnsTrue() throws InterruptedException { Duration processingTime = standardHours(1); when(listener1.handlePotentiallyStuck(any(ListenerContext.class), eq(processingTime))).thenReturn(true); WorkflowInstance instance = executingInstanceBuilder().setType("stuck").setState("start").build(); when(workflowInstances.getWorkflowInstance(instance.id, INCLUDES, null)).thenReturn(instance); Thread thread = new Thread(executor::run); thread.start(); sleep(500); executor.handlePotentiallyStuck(processingTime); thread.join(1000); assertFalse(thread.isAlive(), "Processing thread did not die after interruption"); verify(listener1).handlePotentiallyStuck(any(ListenerContext.class), eq(processingTime)); verify(listener2).handlePotentiallyStuck(any(ListenerContext.class), eq(processingTime)); verify(workflowInstanceDao).updateWorkflowInstanceAfterExecution(update.capture(), action.capture(), childWorkflows.capture(), workflows.capture(), eq(true)); assertThat(action.getValue().type, is(stateExecutionFailed)); assertThat(action.getValue().stateText, containsString("InterruptedException")); }
public void handlePotentiallyStuck(Duration processingTime) { boolean interrupt = false; for (WorkflowExecutorListener listener : executorListeners) { try { if (listener.handlePotentiallyStuck(listenerContext, processingTime)) { interrupt = true; } } catch (Throwable t) { logger.error("Error in " + listener.getClass().getName() + ".handleStuck (" + t.getMessage() + ")", t); } } if (interrupt) { thread.interrupt(); } }
WorkflowStateProcessor implements Runnable { public void handlePotentiallyStuck(Duration processingTime) { boolean interrupt = false; for (WorkflowExecutorListener listener : executorListeners) { try { if (listener.handlePotentiallyStuck(listenerContext, processingTime)) { interrupt = true; } } catch (Throwable t) { logger.error("Error in " + listener.getClass().getName() + ".handleStuck (" + t.getMessage() + ")", t); } } if (interrupt) { thread.interrupt(); } } }
WorkflowStateProcessor implements Runnable { public void handlePotentiallyStuck(Duration processingTime) { boolean interrupt = false; for (WorkflowExecutorListener listener : executorListeners) { try { if (listener.handlePotentiallyStuck(listenerContext, processingTime)) { interrupt = true; } } catch (Throwable t) { logger.error("Error in " + listener.getClass().getName() + ".handleStuck (" + t.getMessage() + ")", t); } } if (interrupt) { thread.interrupt(); } } WorkflowStateProcessor(long instanceId, Supplier<Boolean> shutdownRequested, ObjectStringMapper objectMapper, WorkflowDefinitionService workflowDefinitions,
      WorkflowInstanceService workflowInstances, WorkflowInstanceDao workflowInstanceDao, MaintenanceDao maintenanceDao,
      WorkflowInstancePreProcessor workflowInstancePreProcessor, Environment env,
      Map<Long, WorkflowStateProcessor> processingInstances, WorkflowExecutorListener... executorListeners); }
WorkflowStateProcessor implements Runnable { public void handlePotentiallyStuck(Duration processingTime) { boolean interrupt = false; for (WorkflowExecutorListener listener : executorListeners) { try { if (listener.handlePotentiallyStuck(listenerContext, processingTime)) { interrupt = true; } } catch (Throwable t) { logger.error("Error in " + listener.getClass().getName() + ".handleStuck (" + t.getMessage() + ")", t); } } if (interrupt) { thread.interrupt(); } } WorkflowStateProcessor(long instanceId, Supplier<Boolean> shutdownRequested, ObjectStringMapper objectMapper, WorkflowDefinitionService workflowDefinitions,
      WorkflowInstanceService workflowInstances, WorkflowInstanceDao workflowInstanceDao, MaintenanceDao maintenanceDao,
      WorkflowInstancePreProcessor workflowInstancePreProcessor, Environment env,
      Map<Long, WorkflowStateProcessor> processingInstances, WorkflowExecutorListener... executorListeners); @Override void run(); DateTime getStartTime(); void logPotentiallyStuck(long processingTimeSeconds); void handlePotentiallyStuck(Duration processingTime); }
WorkflowStateProcessor implements Runnable { public void handlePotentiallyStuck(Duration processingTime) { boolean interrupt = false; for (WorkflowExecutorListener listener : executorListeners) { try { if (listener.handlePotentiallyStuck(listenerContext, processingTime)) { interrupt = true; } } catch (Throwable t) { logger.error("Error in " + listener.getClass().getName() + ".handleStuck (" + t.getMessage() + ")", t); } } if (interrupt) { thread.interrupt(); } } WorkflowStateProcessor(long instanceId, Supplier<Boolean> shutdownRequested, ObjectStringMapper objectMapper, WorkflowDefinitionService workflowDefinitions,
      WorkflowInstanceService workflowInstances, WorkflowInstanceDao workflowInstanceDao, MaintenanceDao maintenanceDao,
      WorkflowInstancePreProcessor workflowInstancePreProcessor, Environment env,
      Map<Long, WorkflowStateProcessor> processingInstances, WorkflowExecutorListener... executorListeners); @Override void run(); DateTime getStartTime(); void logPotentiallyStuck(long processingTimeSeconds); void handlePotentiallyStuck(Duration processingTime);  }