DefaultLoadBalancerService implements LoadBalancerService { @VisibleForTesting Observable<Batch<TargetStateBatchable, String>> events() { return engine.events(); } @Inject  DefaultLoadBalancerService(TitusRuntime runtime,
                                      LoadBalancerConfiguration configuration,
                                      LoadBalancerConnector loadBalancerConnector,
                                      LoadBalancerStore loadBalancerStore,
                                      V3JobOperations v3JobOperations,
                                      LoadBalancerJobValidator validator); @VisibleForTesting  DefaultLoadBalancerService(TitusRuntime runtime,
                               LoadBalancerConfiguration configuration,
                               LoadBalancerConnector loadBalancerConnector,
                               LoadBalancerStore loadBalancerStore,
                               LoadBalancerJobOperations loadBalancerJobOperations,
                               LoadBalancerReconciler reconciler,
                               LoadBalancerJobValidator validator,
                               Scheduler scheduler); @Override Observable<String> getJobLoadBalancers(String jobId); @Override Pair<List<JobLoadBalancer>, Pagination> getAllLoadBalancers(Page page); @Override Completable addLoadBalancer(String jobId, String loadBalancerId); @Override Completable removeLoadBalancer(String jobId, String loadBalancerId); @Activator void activate(); @Deactivator void deactivate();  }
@Test public void emptyBatchesAreFilteredOut() { defaultStubs(); LoadBalancerConfiguration configuration = LoadBalancerTests.mockConfiguration(MIN_TIME_IN_QUEUE_MS); DefaultLoadBalancerService service = new DefaultLoadBalancerService( runtime, configuration, client, loadBalancerStore, loadBalancerJobOperations, reconciler, validator, testScheduler); AssertableSubscriber<Batch<TargetStateBatchable, String>> testSubscriber = service.events().test(); testSubscriber.assertNoErrors().assertValueCount(0); verify(client, never()).registerAll(any(), any()); verify(client, never()).deregisterAll(any(), any()); verifyNoReconcilerIgnore(); testScheduler.advanceTimeBy(FLUSH_WAIT_TIME_MS, TimeUnit.MILLISECONDS); testSubscriber.assertNoErrors().assertValueCount(0); verify(client, never()).registerAll(any(), any()); verify(client, never()).deregisterAll(any(), any()); verifyNoReconcilerIgnore(); }