### Question:
RestClusterClient extends ClusterClient<T> implements NewClusterClient { @Override public CompletableFuture<Collection<JobStatusMessage>> listJobs() { return sendRequest(JobsOverviewHeaders.getInstance()) .thenApply( (multipleJobsDetails) -> multipleJobsDetails .getJobs() .stream() .map(detail -> new JobStatusMessage( detail.getJobId(), detail.getJobName(), detail.getStatus(), detail.getStartTime())) .collect(Collectors.toList())); } RestClusterClient(Configuration config, T clusterId);  RestClusterClient(
			Configuration config,
			T clusterId,
			LeaderRetrievalService webMonitorRetrievalService); @VisibleForTesting  RestClusterClient(
			Configuration configuration,
			@Nullable RestClient restClient,
			T clusterId,
			WaitStrategy waitStrategy,
			@Nullable LeaderRetrievalService webMonitorRetrievalService); @Override void shutdown(); @Override JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader); CompletableFuture<JobDetailsInfo> getJobDetails(JobID jobId); @Override CompletableFuture<JobStatus> getJobStatus(JobID jobId); @Override CompletableFuture<JobResult> requestJobResult(@Nonnull JobID jobId); @Override CompletableFuture<JobSubmissionResult> submitJob(@Nonnull JobGraph jobGraph); @Override void stop(JobID jobID); @Override void cancel(JobID jobID); @Override String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<String> triggerSavepoint(
			final JobID jobId,
			final @Nullable String savepointDirectory); @Override Map<String, OptionalFailure<Object>> getAccumulators(final JobID jobID, ClassLoader loader); @Override CompletableFuture<Collection<JobStatusMessage>> listJobs(); @Override T getClusterId(); @Override LeaderConnectionInfo getClusterConnectionInfo(); @Override CompletableFuture<Acknowledge> rescaleJob(JobID jobId, int newParallelism); @Override CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath); @Override void shutDownCluster(); @Override boolean hasUserJarsInClassPath(List<URL> userJarFiles); @Override void waitForClusterToBeReady(); @Override String getWebInterfaceURL(); @Override GetClusterStatusResponse getClusterStatus(); @Override List<String> getNewMessages(); @Override int getMaxSlots(); @VisibleForTesting CompletableFuture<P> sendRequest(M messageHeaders, U messageParameters, R request);  }### Answer:
@Test public void testListJobs() throws Exception { try (TestRestServerEndpoint restServerEndpoint = createRestServerEndpoint(new TestListJobsHandler())) { RestClusterClient<?> restClusterClient = createRestClusterClient(restServerEndpoint.getServerAddress().getPort()); try { CompletableFuture<Collection<JobStatusMessage>> jobDetailsFuture = restClusterClient.listJobs(); Collection<JobStatusMessage> jobDetails = jobDetailsFuture.get(); Iterator<JobStatusMessage> jobDetailsIterator = jobDetails.iterator(); JobStatusMessage job1 = jobDetailsIterator.next(); JobStatusMessage job2 = jobDetailsIterator.next(); Assert.assertNotEquals("The job status should not be equal.", job1.getJobState(), job2.getJobState()); } finally { restClusterClient.shutdown(); } } }