@Test(timeout = TEST_TIMEOUT_MS) public void updatesAreIgnoredWhileCooldownIsActive() { long cooldownPeriodMs = 5 * delayMs; List<Task> tasks = LoadBalancerTests.buildTasksStarted(5, jobId); JobLoadBalancer jobLoadBalancer = new JobLoadBalancer(jobId, loadBalancerId); JobLoadBalancerState association = new JobLoadBalancerState(jobLoadBalancer, State.ASSOCIATED); when(v3JobOperations.getTasks(jobId)).thenReturn(tasks); store.addOrUpdateLoadBalancer(association.getJobLoadBalancer(), association.getState()).await(); for (Task task : tasks) { String ipAddress = task.getTaskContext().get(TaskAttributes.TASK_ATTRIBUTES_CONTAINER_IP); LoadBalancerTarget target = new LoadBalancerTarget(loadBalancerId, task.getId(), ipAddress); reconciler.activateCooldownFor(target, cooldownPeriodMs, TimeUnit.MILLISECONDS); } testScheduler.triggerActions(); subscriber.assertNotCompleted().assertNoValues(); awaitReconciliationRuns(1); subscriber.assertNotCompleted().assertNoValues(); awaitReconciliationRuns(4); subscriber.assertNotCompleted().assertValueCount(5); subscriber.getOnNextEvents().forEach(update -> { assertThat(update.getState()).isEqualTo(LoadBalancerTarget.State.REGISTERED); assertThat(update.getPriority()).isEqualTo(Priority.LOW); assertThat(update.getLoadBalancerId()).isEqualTo(loadBalancerId); }); awaitReconciliationRuns(1); subscriber.assertNotCompleted().assertValueCount(10); }
@Override public void activateCooldownFor(LoadBalancerTarget target, long period, TimeUnit unit) { Duration periodDuration = Duration.ofMillis(unit.toMillis(period)); logger.debug("Setting a cooldown of {} for target {}", periodDuration, target); Instant untilWhen = Instant.ofEpochMilli(scheduler.now()).plus(periodDuration); ignored.put(target, untilWhen); }
DefaultLoadBalancerReconciler implements LoadBalancerReconciler { @Override public void activateCooldownFor(LoadBalancerTarget target, long period, TimeUnit unit) { Duration periodDuration = Duration.ofMillis(unit.toMillis(period)); logger.debug("Setting a cooldown of {} for target {}", periodDuration, target); Instant untilWhen = Instant.ofEpochMilli(scheduler.now()).plus(periodDuration); ignored.put(target, untilWhen); } }
DefaultLoadBalancerReconciler implements LoadBalancerReconciler { @Override public void activateCooldownFor(LoadBalancerTarget target, long period, TimeUnit unit) { Duration periodDuration = Duration.ofMillis(unit.toMillis(period)); logger.debug("Setting a cooldown of {} for target {}", periodDuration, target); Instant untilWhen = Instant.ofEpochMilli(scheduler.now()).plus(periodDuration); ignored.put(target, untilWhen); } DefaultLoadBalancerReconciler(LoadBalancerConfiguration configuration,
                                  LoadBalancerStore store,
                                  LoadBalancerConnector connector,
                                  LoadBalancerJobOperations loadBalancerJobOperations,
                                  Registry registry,
                                  Scheduler scheduler);   DefaultLoadBalancerReconciler(LoadBalancerConfiguration configuration,
                                  LoadBalancerStore store,
                                  LoadBalancerConnector connector,
                                  LoadBalancerJobOperations loadBalancerJobOperations,
                                  Runnable afterReconciliation,
                                  Registry registry,
                                  Scheduler scheduler); }
DefaultLoadBalancerReconciler implements LoadBalancerReconciler { @Override public void activateCooldownFor(LoadBalancerTarget target, long period, TimeUnit unit) { Duration periodDuration = Duration.ofMillis(unit.toMillis(period)); logger.debug("Setting a cooldown of {} for target {}", periodDuration, target); Instant untilWhen = Instant.ofEpochMilli(scheduler.now()).plus(periodDuration); ignored.put(target, untilWhen); } DefaultLoadBalancerReconciler(LoadBalancerConfiguration configuration,
                                  LoadBalancerStore store,
                                  LoadBalancerConnector connector,
                                  LoadBalancerJobOperations loadBalancerJobOperations,
                                  Registry registry,
                                  Scheduler scheduler);   DefaultLoadBalancerReconciler(LoadBalancerConfiguration configuration,
                                  LoadBalancerStore store,
                                  LoadBalancerConnector connector,
                                  LoadBalancerJobOperations loadBalancerJobOperations,
                                  Runnable afterReconciliation,
                                  Registry registry,
                                  Scheduler scheduler); @Override void activateCooldownFor(LoadBalancerTarget target, long period, TimeUnit unit); @Override void shutdown(); @Override Observable<TargetStateBatchable> events(); }
DefaultLoadBalancerReconciler implements LoadBalancerReconciler { @Override public void activateCooldownFor(LoadBalancerTarget target, long period, TimeUnit unit) { Duration periodDuration = Duration.ofMillis(unit.toMillis(period)); logger.debug("Setting a cooldown of {} for target {}", periodDuration, target); Instant untilWhen = Instant.ofEpochMilli(scheduler.now()).plus(periodDuration); ignored.put(target, untilWhen); } DefaultLoadBalancerReconciler(LoadBalancerConfiguration configuration,
                                  LoadBalancerStore store,
                                  LoadBalancerConnector connector,
                                  LoadBalancerJobOperations loadBalancerJobOperations,
                                  Registry registry,
                                  Scheduler scheduler);   DefaultLoadBalancerReconciler(LoadBalancerConfiguration configuration,
                                  LoadBalancerStore store,
                                  LoadBalancerConnector connector,
                                  LoadBalancerJobOperations loadBalancerJobOperations,
                                  Runnable afterReconciliation,
                                  Registry registry,
                                  Scheduler scheduler); @Override void activateCooldownFor(LoadBalancerTarget target, long period, TimeUnit unit); @Override void shutdown(); @Override Observable<TargetStateBatchable> events();  }