@Test public void noTerminationsOnDataStaleness() { TitusRuntime titusRuntime = mock(TitusRuntime.class); when(titusRuntime.getRegistry()).thenReturn(new DefaultRegistry()); RelocationConfiguration configuration = mock(RelocationConfiguration.class); when(configuration.getBadNodeConditionPattern()).thenReturn(".*Problem"); when(configuration.isTaskTerminationOnBadNodeConditionEnabled()).thenReturn(true); when(configuration.getDataStalenessThresholdMs()).thenReturn(8000L); NodeDataResolver nodeDataResolver = mock(NodeDataResolver.class); when(nodeDataResolver.getStalenessMs()).thenReturn(5L); JobDataReplicator jobDataReplicator = mock(JobDataReplicator.class); when(jobDataReplicator.getStalenessMs()).thenReturn(10L); ReadOnlyJobOperations readOnlyJobOperations = mock(ReadOnlyJobOperations.class); JobManagementClient jobManagementClient = mock(JobManagementClient.class); Set<String> terminatedTaskIds = new HashSet<>(); when(jobManagementClient.killTask(anyString(), anyBoolean(), any())).thenAnswer(invocation -> { String taskIdToBeTerminated = invocation.getArgument(0); terminatedTaskIds.add(taskIdToBeTerminated); return Mono.empty(); }); DefaultNodeConditionController nodeConditionCtrl = new DefaultNodeConditionController(configuration, nodeDataResolver, jobDataReplicator, readOnlyJobOperations, jobManagementClient, titusRuntime); ExecutionContext executionContext = ExecutionContext.newBuilder().withIteration(ExecutionId.initial()).build(); StepVerifier.create(nodeConditionCtrl.handleNodesWithBadCondition(executionContext)) .verifyComplete(); assertThat(terminatedTaskIds).isEmpty(); }
@VisibleForTesting Mono<Void> handleNodesWithBadCondition(ExecutionContext executionContext) { int iterationCount = executionContext.getExecutionId().getTotal(); logger.debug("Starting node condition controller iteration {} ...", iterationCount); if (hasStaleData()) { logger.info("Stale data. Skipping the node condition control loop iteration- {} ", iterationCount); return Mono.empty(); } return handleNodesWithBadCondition(); }
DefaultNodeConditionController implements NodeConditionController { @VisibleForTesting Mono<Void> handleNodesWithBadCondition(ExecutionContext executionContext) { int iterationCount = executionContext.getExecutionId().getTotal(); logger.debug("Starting node condition controller iteration {} ...", iterationCount); if (hasStaleData()) { logger.info("Stale data. Skipping the node condition control loop iteration- {} ", iterationCount); return Mono.empty(); } return handleNodesWithBadCondition(); } }
DefaultNodeConditionController implements NodeConditionController { @VisibleForTesting Mono<Void> handleNodesWithBadCondition(ExecutionContext executionContext) { int iterationCount = executionContext.getExecutionId().getTotal(); logger.debug("Starting node condition controller iteration {} ...", iterationCount); if (hasStaleData()) { logger.info("Stale data. Skipping the node condition control loop iteration- {} ", iterationCount); return Mono.empty(); } return handleNodesWithBadCondition(); } DefaultNodeConditionController(RelocationConfiguration relocationConfiguration,
                                          NodeDataResolver nodeDataResolver,
                                          JobDataReplicator jobDataReplicator,
                                          ReadOnlyJobOperations jobOperations,
                                          JobManagementClient jobManagementClient,
                                          TitusRuntime titusRuntime); }
DefaultNodeConditionController implements NodeConditionController { @VisibleForTesting Mono<Void> handleNodesWithBadCondition(ExecutionContext executionContext) { int iterationCount = executionContext.getExecutionId().getTotal(); logger.debug("Starting node condition controller iteration {} ...", iterationCount); if (hasStaleData()) { logger.info("Stale data. Skipping the node condition control loop iteration- {} ", iterationCount); return Mono.empty(); } return handleNodesWithBadCondition(); } DefaultNodeConditionController(RelocationConfiguration relocationConfiguration,
                                          NodeDataResolver nodeDataResolver,
                                          JobDataReplicator jobDataReplicator,
                                          ReadOnlyJobOperations jobOperations,
                                          JobManagementClient jobManagementClient,
                                          TitusRuntime titusRuntime); @Override void activate(); @Override void deactivate(); Mono<Void> handleNodesWithBadCondition(); }
DefaultNodeConditionController implements NodeConditionController { @VisibleForTesting Mono<Void> handleNodesWithBadCondition(ExecutionContext executionContext) { int iterationCount = executionContext.getExecutionId().getTotal(); logger.debug("Starting node condition controller iteration {} ...", iterationCount); if (hasStaleData()) { logger.info("Stale data. Skipping the node condition control loop iteration- {} ", iterationCount); return Mono.empty(); } return handleNodesWithBadCondition(); } DefaultNodeConditionController(RelocationConfiguration relocationConfiguration,
                                          NodeDataResolver nodeDataResolver,
                                          JobDataReplicator jobDataReplicator,
                                          ReadOnlyJobOperations jobOperations,
                                          JobManagementClient jobManagementClient,
                                          TitusRuntime titusRuntime); @Override void activate(); @Override void deactivate(); Mono<Void> handleNodesWithBadCondition();  }