DefaultLoadBalancerReconciler implements LoadBalancerReconciler { private List<TargetStateBatchable> withState(Instant instant, Collection<LoadBalancerTarget> targets, State state) { return targets.stream() .map(target -> new TargetStateBatchable(Priority.LOW, instant, new LoadBalancerTargetState(target, state))) .collect(Collectors.toList()); } 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();  }
@Test public void orphanLoadBalancerAssociationsAreSetAsDissociatedAndRemoved() { List<Task> tasks = LoadBalancerTests.buildTasksStarted(5, jobId); JobLoadBalancer jobLoadBalancer = new JobLoadBalancer(jobId, loadBalancerId); when(v3JobOperations.getTasks(jobId)).thenReturn(tasks); reset(connector); OngoingStubbing<Single<LoadBalancer>> ongoingStubbing = when(connector.getLoadBalancer(loadBalancerId)) .thenReturn(Single.just(new LoadBalancer( loadBalancerId, LoadBalancer.State.ACTIVE, CollectionsExt.asSet("1.1.1.1", "2.2.2.2", "3.3.3.3", "4.4.4.4", "5.5.5.5") ))); assertThat(store.addOrUpdateLoadBalancer(jobLoadBalancer, State.ASSOCIATED) .await(5, TimeUnit.SECONDS)).isTrue(); store.addOrUpdateTargets(tasks.stream() .map(task -> new LoadBalancerTargetState( new LoadBalancerTarget(loadBalancerId, task.getId(), task.getTaskContext().get(TaskAttributes.TASK_ATTRIBUTES_CONTAINER_IP)), LoadBalancerTarget.State.REGISTERED)) .collect(Collectors.toList()) ).block(); testScheduler.triggerActions(); subscriber.assertNotCompleted().assertNoValues(); ongoingStubbing.thenReturn(Single.just( new LoadBalancer(loadBalancerId, LoadBalancer.State.REMOVED, Collections.emptySet()) )); awaitReconciliationRuns(2); subscriber.awaitValueCount(5, TEST_TIMEOUT_MS / 2, TimeUnit.MILLISECONDS) .assertNoErrors(); assertThat(subscriber.getOnNextEvents()).allMatch(update -> update.getState().equals(LoadBalancerTarget.State.DEREGISTERED)); List<LoadBalancerTargetState> currentTargets = store.getLoadBalancerTargets(loadBalancerId).collectList().block(); assertThat(currentTargets).isNotNull(); store.addOrUpdateTargets(currentTargets.stream() .map(targetState -> targetState.getLoadBalancerTarget().withState(LoadBalancerTarget.State.DEREGISTERED)) .collect(Collectors.toList()) ).block(); awaitReconciliationRuns(1); assertThat(store.getAssociations()).containsOnly(new JobLoadBalancerState(jobLoadBalancer, State.DISSOCIATED)); awaitReconciliationRuns(2); assertThat(store.getLoadBalancerTargets(loadBalancerId).collectList().block()).isEmpty(); assertThat(store.getAssociations()).isEmpty(); assertThat(store.getAssociatedLoadBalancersSetForJob(jobId)).isEmpty(); }