RestClusterClient extends ClusterClient { @Override public CompletableFuture<Collection<JobStatusMessage>> listJobs() throws Exception { CurrentJobsOverviewHandlerHeaders headers = CurrentJobsOverviewHandlerHeaders.getInstance(); CompletableFuture<MultipleJobsDetails> jobDetailsFuture = restClient.sendRequest( restClusterClientConfiguration.getRestServerAddress(), restClusterClientConfiguration.getRestServerPort(), headers ); return jobDetailsFuture .thenApply(details -> { Collection<JobStatusMessage> flattenedDetails = new ArrayList<>(); details.getRunning().forEach(detail -> flattenedDetails.add(new JobStatusMessage(detail.getJobId(), detail.getJobName(), detail.getStatus(), detail.getStartTime()))); details.getFinished().forEach(detail -> flattenedDetails.add(new JobStatusMessage(detail.getJobId(), detail.getJobName(), detail.getStatus(), detail.getStartTime()))); return flattenedDetails; }); } RestClusterClient(Configuration config);  RestClusterClient(Configuration config, RestClusterClientConfiguration configuration); @Override void shutdown(); @Override void stop(JobID jobID); @Override void cancel(JobID jobID); @Override String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<Collection<JobStatusMessage>> listJobs(); @Override String getClusterIdentifier(); @Override boolean hasUserJarsInClassPath(List<URL> userJarFiles); @Override void waitForClusterToBeReady(); @Override String getWebInterfaceURL(); @Override GetClusterStatusResponse getClusterStatus(); @Override int getMaxSlots();  }
@Test public void testListJobs() throws Exception { Configuration config = new Configuration(); config.setString(JobManagerOptions.ADDRESS, "localhost"); RestServerEndpointConfiguration rsec = RestServerEndpointConfiguration.fromConfiguration(config); TestListJobsHandler listJobsHandler = new TestListJobsHandler(); RestServerEndpoint rse = new RestServerEndpoint(rsec) { @Override protected Collection<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(CompletableFuture<String> restAddressFuture) { Collection<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = new ArrayList<>(); handlers.add(Tuple2.of(listJobsHandler.getMessageHeaders(), listJobsHandler)); return handlers; } }; RestClusterClient rcc = new RestClusterClient(config); try { rse.start(); { CompletableFuture<Collection<JobStatusMessage>> jobDetailsFuture = rcc.listJobs(); Collection<JobStatusMessage> jobDetails = jobDetailsFuture.get(); Iterator<JobStatusMessage> jobDetailsIterator = jobDetails.iterator(); JobStatusMessage job1 = jobDetailsIterator.next(); JobStatusMessage job2 = jobDetailsIterator.next(); Assert.assertNotEquals("The job statues should not be equal.", job1.getJobState(), job2.getJobState()); } } finally { rcc.shutdown(); rse.shutdown(Time.seconds(5)); }}