@Test
public void observeJob() {

    String cellOneJobId = UUID.randomUUID().toString();
    String cellTwoJobId = UUID.randomUUID().toString();
    cellOne.getServiceRegistry().addService(new CellWithJobStream(cellOneJobId, cellOneUpdates.serialize()));
    cellTwo.getServiceRegistry().addService(new CellWithJobStream(cellTwoJobId, cellTwoUpdates.serialize()));
    AssertableSubscriber<JobChangeNotification> subscriber1 = service.observeJob(cellOneJobId, UNDEFINED_CALL_METADATA).test();
    AssertableSubscriber<JobChangeNotification> subscriber2 = service.observeJob(cellTwoJobId, UNDEFINED_CALL_METADATA).test();
    cellOneUpdates.onNext(toNotification(Job.newBuilder().setId(cellOneJobId).setStatus(ACCEPTED_STATE).build()));
    cellOneUpdates.onNext(toNotification(Job.newBuilder().setId(cellOneJobId).setStatus(KILL_INITIATED_STATE).build()));
    cellOneUpdates.onNext(toNotification(Job.newBuilder().setId(cellOneJobId).setStatus(FINISHED_STATE).build()));
    cellOneUpdates.onNext(toNotification(Job.newBuilder().setId(cellOneJobId).setStatus(ACCEPTED_STATE).build()));
    subscriber1.awaitValueCount(3, 5, TimeUnit.SECONDS);
    subscriber1.assertNoErrors();
    subscriber1.assertNotCompleted();
    assertThat(subscriber1.isUnsubscribed()).isFalse();
    subscriber2.assertNoErrors();
    subscriber2.assertNoValues();
    subscriber2.assertNotCompleted();
    cellTwoUpdates.onNext(toNotification(Task.newBuilder()
            .setId(cellTwoJobId + "-task1").setJobId(cellTwoJobId)
            .build())
    );
    subscriber2.awaitValueCount(1, 1, TimeUnit.SECONDS);
    subscriber2.assertNoErrors();
    subscriber2.assertNotCompleted();
    cellOneUpdates.onCompleted();
    subscriber1.awaitTerminalEvent(1, TimeUnit.SECONDS);
    assertThat(subscriber1.getOnErrorEvents()).isEmpty();
    assertThat(subscriber1.isUnsubscribed()).isTrue();
    assertThat(subscriber1.getCompletions()).isEqualTo(1);
}