@Test
public void observeJobsStopsWhenAnyClientsTerminate() {

    cellOne.getServiceRegistry().addService(new CellWithFixedJobsService(Collections.emptyList(), cellOneUpdates.serialize()));
    cellTwo.getServiceRegistry().addService(new CellWithFixedJobsService(Collections.emptyList(), cellTwoUpdates.serialize()));
    final AssertableSubscriber<JobChangeNotification> testSubscriber = service.observeJobs(ObserveJobsQuery.getDefaultInstance(), UNDEFINED_CALL_METADATA).test();
    final JobChangeNotification cellOneUpdate = toNotification(Job.newBuilder().setId("cell-1-job-100").setStatus(ACCEPTED_STATE).build());
    final JobChangeNotification cellTwoUpdate = toNotification(Job.newBuilder().setId("cell-2-job-200").setStatus(ACCEPTED_STATE).build());
    cellOneUpdates.onNext(cellOneUpdate);
    cellTwoUpdates.onNext(cellTwoUpdate);
    testSubscriber.awaitValueCount(2, 1, TimeUnit.SECONDS);
    assertThat(testSubscriber.getOnErrorEvents()).isEmpty();
    assertThat(testSubscriber.isUnsubscribed()).isFalse();
    assertThat(testSubscriber.getCompletions()).isEqualTo(0);
    // a client finishes
    cellTwoUpdates.onCompleted();
    testSubscriber.awaitTerminalEvent(1, TimeUnit.SECONDS);
    assertThat(testSubscriber.getOnErrorEvents()).isEmpty();
    assertThat(testSubscriber.isUnsubscribed()).isTrue();
    assertThat(testSubscriber.getCompletions()).isEqualTo(1);
}