@Test public void testScaleStageRequestRetriesNewMaster() throws InterruptedException { MasterMonitor mockMasterMonitor = mock(MasterMonitor.class); final BehaviorSubject<MasterDescription> mdSubject = BehaviorSubject.create(); when(mockMasterMonitor.getMasterObservable()).thenReturn(mdSubject); MantisMasterClientApi mantisMasterClientApi = new MantisMasterClientApi(mockMasterMonitor); final int oldMasterPort = port.incrementAndGet(); final int newMasterPort = port.incrementAndGet(); final String jobId = "test-job-id"; final int stageNum = 1; final int numWorkers = 2; final String reason = "test reason"; mdSubject.onNext(new MasterDescription("localhost", "127.0.0.1", oldMasterPort, 7090, 7091, "status", 8900, System.currentTimeMillis())); final CountDownLatch retryLatch = new CountDownLatch(3); final Func1<Observable<? extends Throwable>, Observable<?>> retryLogic = new Func1<Observable<? extends Throwable>, Observable<?>>() { @Override public Observable<?> call(Observable<? extends Throwable> attempts) { return attempts .zipWith(Observable.range(1, 5), new Func2<Throwable, Integer, Integer>() { @Override public Integer call(Throwable t1, Integer integer) { return integer; } }) .flatMap(new Func1<Integer, Observable<?>>() { @Override public Observable<?> call(Integer retryCount) { logger.info(retryCount + " retrying conx after sleeping for 250ms"); if (retryCount == 2) { Schedulers.newThread().createWorker().schedule(new Action0() { @Override public void call() { final HttpServer<String, String> httpServer = createHttpServer(newMasterPort); startedServers.add(httpServer); httpServer.start(); } }); } if (retryCount == 3) { mdSubject.onNext(new MasterDescription("localhost", "127.0.0.1", newMasterPort, 7090, 7091, "status", 8900, System.currentTimeMillis())); } retryLatch.countDown(); return Observable.timer(250, TimeUnit.MILLISECONDS); } }); } }; final Observable<Boolean> resultObs = mantisMasterClientApi.scaleJobStage(jobId, stageNum, numWorkers, reason) .retryWhen(retryLogic); final CountDownLatch completedLatch = new CountDownLatch(1); resultObs .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { fail("got unexpected error" + throwable.getMessage()); } }) .doOnCompleted(new Action0() { @Override public void call() { completedLatch.countDown(); } }).subscribe(); assertTrue(retryLatch.await(5, TimeUnit.SECONDS)); assertTrue(completedLatch.await(5, TimeUnit.SECONDS)); }
public Observable<Boolean> scaleJobStage(final String jobId, final int stageNum, final int numWorkers, final String reason) { return masterMonitor .getMasterObservable() .filter(md -> md != null) .take(1) .flatMap((Func1<MasterDescription, Observable<Boolean>>) md -> { final StageScaleRequest stageScaleRequest = new StageScaleRequest(jobId, stageNum, numWorkers, reason); try { return submitPostRequest(toUri(md, API_JOB_STAGE_SCALE), objectMapper.writeValueAsString(stageScaleRequest)) .map(s -> { logger.info("POST to scale stage returned status: {}", s); return s.codeClass().equals(HttpStatusClass.SUCCESS); }); } catch (JsonProcessingException e) { logger.error("failed to serialize stage scale request {} to json", stageScaleRequest); return Observable.error(e); } }); }
MantisMasterClientApi { public Observable<Boolean> scaleJobStage(final String jobId, final int stageNum, final int numWorkers, final String reason) { return masterMonitor .getMasterObservable() .filter(md -> md != null) .take(1) .flatMap((Func1<MasterDescription, Observable<Boolean>>) md -> { final StageScaleRequest stageScaleRequest = new StageScaleRequest(jobId, stageNum, numWorkers, reason); try { return submitPostRequest(toUri(md, API_JOB_STAGE_SCALE), objectMapper.writeValueAsString(stageScaleRequest)) .map(s -> { logger.info("POST to scale stage returned status: {}", s); return s.codeClass().equals(HttpStatusClass.SUCCESS); }); } catch (JsonProcessingException e) { logger.error("failed to serialize stage scale request {} to json", stageScaleRequest); return Observable.error(e); } }); } }
MantisMasterClientApi { public Observable<Boolean> scaleJobStage(final String jobId, final int stageNum, final int numWorkers, final String reason) { return masterMonitor .getMasterObservable() .filter(md -> md != null) .take(1) .flatMap((Func1<MasterDescription, Observable<Boolean>>) md -> { final StageScaleRequest stageScaleRequest = new StageScaleRequest(jobId, stageNum, numWorkers, reason); try { return submitPostRequest(toUri(md, API_JOB_STAGE_SCALE), objectMapper.writeValueAsString(stageScaleRequest)) .map(s -> { logger.info("POST to scale stage returned status: {}", s); return s.codeClass().equals(HttpStatusClass.SUCCESS); }); } catch (JsonProcessingException e) { logger.error("failed to serialize stage scale request {} to json", stageScaleRequest); return Observable.error(e); } }); } MantisMasterClientApi(MasterMonitor masterMonitor); }
MantisMasterClientApi { public Observable<Boolean> scaleJobStage(final String jobId, final int stageNum, final int numWorkers, final String reason) { return masterMonitor .getMasterObservable() .filter(md -> md != null) .take(1) .flatMap((Func1<MasterDescription, Observable<Boolean>>) md -> { final StageScaleRequest stageScaleRequest = new StageScaleRequest(jobId, stageNum, numWorkers, reason); try { return submitPostRequest(toUri(md, API_JOB_STAGE_SCALE), objectMapper.writeValueAsString(stageScaleRequest)) .map(s -> { logger.info("POST to scale stage returned status: {}", s); return s.codeClass().equals(HttpStatusClass.SUCCESS); }); } catch (JsonProcessingException e) { logger.error("failed to serialize stage scale request {} to json", stageScaleRequest); return Observable.error(e); } }); } MantisMasterClientApi(MasterMonitor masterMonitor); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final SchedulingInfo schedulingInfo); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo,
                                                   final WorkerMigrationConfig migrationConfig); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo,
                                                   final boolean readyForJobMaster); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo,
                                                   final boolean readyForJobMaster,
                                                   final WorkerMigrationConfig migrationConfig); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo,
                                                   final boolean readyForJobMaster,
                                                   final WorkerMigrationConfig migrationConfig,
                                                   final List<Label> labels); Observable<JobSubmitResponse> submitJob(final String submitJobRequestJson); Observable<Void> killJob(final String jobId); Observable<Void> killJob(final String jobId, final String user, final String reason); Observable<Boolean> scaleJobStage(final String jobId, final int stageNum, final int numWorkers,
                                             final String reason); Observable<Boolean> resubmitJobWorker(final String jobId, final String user, final int workerNum,
                                                 final String reason); Observable<Boolean> namedJobExists(final String jobName); Observable<Integer> getSinkStageNum(final String jobId); Observable<String> getJobsOfNamedJob(final String jobName, final MantisJobState.MetaState state); Observable<Boolean> jobIdExists(final String jobId); Observable<String> getJobStatusObservable(final String jobId); Observable<JobSchedulingInfo> schedulingChanges(final String jobId); Observable<NamedJobInfo> namedJobInfo(final String jobName); Observable<JobAssignmentResult> assignmentResults(String jobId); }
MantisMasterClientApi { public Observable<Boolean> scaleJobStage(final String jobId, final int stageNum, final int numWorkers, final String reason) { return masterMonitor .getMasterObservable() .filter(md -> md != null) .take(1) .flatMap((Func1<MasterDescription, Observable<Boolean>>) md -> { final StageScaleRequest stageScaleRequest = new StageScaleRequest(jobId, stageNum, numWorkers, reason); try { return submitPostRequest(toUri(md, API_JOB_STAGE_SCALE), objectMapper.writeValueAsString(stageScaleRequest)) .map(s -> { logger.info("POST to scale stage returned status: {}", s); return s.codeClass().equals(HttpStatusClass.SUCCESS); }); } catch (JsonProcessingException e) { logger.error("failed to serialize stage scale request {} to json", stageScaleRequest); return Observable.error(e); } }); } MantisMasterClientApi(MasterMonitor masterMonitor); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final SchedulingInfo schedulingInfo); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo,
                                                   final WorkerMigrationConfig migrationConfig); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo,
                                                   final boolean readyForJobMaster); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo,
                                                   final boolean readyForJobMaster,
                                                   final WorkerMigrationConfig migrationConfig); Observable<JobSubmitResponse> submitJob(final String name, final String version,
                                                   final List<Parameter> parameters,
                                                   final JobSla jobSla,
                                                   final long subscriptionTimeoutSecs,
                                                   final SchedulingInfo schedulingInfo,
                                                   final boolean readyForJobMaster,
                                                   final WorkerMigrationConfig migrationConfig,
                                                   final List<Label> labels); Observable<JobSubmitResponse> submitJob(final String submitJobRequestJson); Observable<Void> killJob(final String jobId); Observable<Void> killJob(final String jobId, final String user, final String reason); Observable<Boolean> scaleJobStage(final String jobId, final int stageNum, final int numWorkers,
                                             final String reason); Observable<Boolean> resubmitJobWorker(final String jobId, final String user, final int workerNum,
                                                 final String reason); Observable<Boolean> namedJobExists(final String jobName); Observable<Integer> getSinkStageNum(final String jobId); Observable<String> getJobsOfNamedJob(final String jobName, final MantisJobState.MetaState state); Observable<Boolean> jobIdExists(final String jobId); Observable<String> getJobStatusObservable(final String jobId); Observable<JobSchedulingInfo> schedulingChanges(final String jobId); Observable<NamedJobInfo> namedJobInfo(final String jobName); Observable<JobAssignmentResult> assignmentResults(String jobId);  }