@Test
public void observeJobsWaitsForAllMarkers() {

    final List<Job> cellOneSnapshot = Arrays.asList(
            Job.newBuilder().setId("cell-1-job-1").setStatus(ACCEPTED_STATE).build(),
            Job.newBuilder().setId("cell-1-job-2").setStatus(ACCEPTED_STATE).build(),
            Job.newBuilder().setId("cell-1-job-3").setStatus(ACCEPTED_STATE).build()
    );
    final List<Job> cellTwoSnapshot = Arrays.asList(
            Job.newBuilder().setId("cell-2-job-1").setStatus(ACCEPTED_STATE).build(),
            Job.newBuilder().setId("cell-2-job-2").setStatus(ACCEPTED_STATE).build(),
            Job.newBuilder().setId("cell-2-job-3").setStatus(ACCEPTED_STATE).build()
    );
    cellOne.getServiceRegistry().addService(new CellWithFixedJobsService(cellOneSnapshot, cellOneUpdates.serialize()));
    cellTwo.getServiceRegistry().addService(new CellWithFixedJobsService(cellTwoSnapshot, cellTwoUpdates.serialize()));
    final AssertableSubscriber<JobChangeNotification> testSubscriber = service.observeJobs(ObserveJobsQuery.getDefaultInstance(), UNDEFINED_CALL_METADATA).test();
    List<JobChangeNotification> expected = Stream.concat(
            cellOneSnapshot.stream().map(this::toNotification).map(this::withStackName),
            cellTwoSnapshot.stream().map(this::toNotification).map(this::withStackName)
    ).collect(Collectors.toList());
    // single marker for all cells
    final JobChangeNotification mergedMarker = JobChangeNotification.newBuilder().setSnapshotEnd(SnapshotEnd.newBuilder()).build();
    expected.add(mergedMarker);
    testSubscriber.awaitValueCount(7, 1, TimeUnit.SECONDS);
    List<JobChangeNotification> onNextEvents = testSubscriber.getOnNextEvents();
    assertThat(onNextEvents).last().isEqualTo(mergedMarker);
    assertThat(onNextEvents).containsExactlyInAnyOrder(expected.toArray(new JobChangeNotification[expected.size()]));
    // more updates are flowing
    final JobChangeNotification cellOneUpdate = toNotification(Job.newBuilder().setId("cell-1-job-10").setStatus(ACCEPTED_STATE).build());
    final JobChangeNotification cellTwoUpdate = toNotification(Job.newBuilder().setId("cell-2-job-10").setStatus(ACCEPTED_STATE).build());
    cellOneUpdates.onNext(cellOneUpdate);
    cellTwoUpdates.onNext(cellTwoUpdate);
    testSubscriber.awaitValueCount(9, 1, TimeUnit.SECONDS);
    onNextEvents = testSubscriber.getOnNextEvents();
    assertThat(onNextEvents).last().isNotEqualTo(mergedMarker);
    assertThat(onNextEvents).contains(withStackName(cellOneUpdate), withStackName(cellTwoUpdate));
}