@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();
    // all targets were previously registered by us
    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();
    // load balancer was removed outside of Titus
    ongoingStubbing.thenReturn(Single.just(
            new LoadBalancer(loadBalancerId, LoadBalancer.State.REMOVED, Collections.emptySet())
    ));
    // Let a few iterations run so all phases can be executed:
    // 1. mark as orphan
    // 2. update targets as DEREGISTERED
    awaitReconciliationRuns(2);
    subscriber.awaitValueCount(5, TEST_TIMEOUT_MS / 2, TimeUnit.MILLISECONDS)
            .assertNoErrors();
    assertThat(subscriber.getOnNextEvents()).allMatch(update -> update.getState().equals(LoadBalancerTarget.State.DEREGISTERED));
    // simulate all targets got 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();
    // 3. update orphan as Dissociated
    awaitReconciliationRuns(1);
    assertThat(store.getAssociations()).containsOnly(new JobLoadBalancerState(jobLoadBalancer, State.DISSOCIATED));
    // 4. sweep all targets
    // 5. sweep all Dissociated
    awaitReconciliationRuns(2);
    assertThat(store.getLoadBalancerTargets(loadBalancerId).collectList().block()).isEmpty();
    assertThat(store.getAssociations()).isEmpty();
    assertThat(store.getAssociatedLoadBalancersSetForJob(jobId)).isEmpty();
}