@Test
public void checkV3LiveStreamTargetUpdates() throws Exception {

    String jobIdOne = UUID.randomUUID().toString();
    String jobIdTwo = UUID.randomUUID().toString();
    InMemoryPolicyStore policyStore = new InMemoryPolicyStore();
    V3JobOperations v3JobOperations = mockV3Operations(jobIdOne, jobIdTwo);
    AppScaleClientWithScalingPolicyConstraints appScalingClient = new AppScaleClientWithScalingPolicyConstraints();
    DefaultAppScaleManager appScaleManager = new DefaultAppScaleManager(policyStore,
            new AutoScalingPolicyTests.MockAlarmClient(),
            appScalingClient,
            v3JobOperations,
            new DefaultRegistry(),
            AutoScalingPolicyTests.mockAppScaleManagerConfiguration(),
            Schedulers.immediate(),
            mock(TitusRuntime.class));
    List<String> refIds = submitTwoJobs(appScaleManager, jobIdOne, jobIdTwo, policyStore);
    Assertions.assertThat(refIds.size()).isEqualTo(2);
    CountDownLatch latch = new CountDownLatch(1);
    Observable<String> jobIdTargetUpdates = appScaleManager.v3LiveStreamTargetUpdates();
    List<String> targetsUpdated = new ArrayList<>();
    jobIdTargetUpdates.subscribe(targetUpdated -> {
                log.info("Got ScalableTarget to be updated {}", targetUpdated);
                Assertions.assertThat(targetUpdated).isEqualTo(jobIdTwo);
                targetsUpdated.add(targetUpdated);
                latch.countDown();
            },
            e -> log.error("Error in v2 live stream for scalable target update", e),
            () -> log.info("Completed"));
    latch.await(60, TimeUnit.SECONDS);
    AutoScalingPolicyTests.waitForCondition(() -> {
        JobScalingConstraints jpc = appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdTwo);
        return jpc != null && jpc.getMinCapacity() == 5 && jpc.getMaxCapacity() == 15;
    });
    Assertions.assertThat(targetsUpdated.size()).isEqualTo(1);
    Assertions.assertThat(appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdOne).getMinCapacity()).isEqualTo(1);
    Assertions.assertThat(appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdOne).getMaxCapacity()).isEqualTo(10);
    Assertions.assertThat(appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdTwo).getMinCapacity()).isEqualTo(5);
    Assertions.assertThat(appScalingClient.getJobScalingPolicyConstraintsForJob(jobIdTwo).getMaxCapacity()).isEqualTo(15);
}