TitusClientImpl implements TitusClient { @Override public Mono<Task> getTask(String taskId) { logger.debug("Getting Task information about taskId {}", taskId); return Mono.create(sink -> attachCallerId(jobManagementService, CLIENT_ID) .findTask(TaskId.newBuilder().setId(taskId).build(), new StreamObserver<Task>() { @Override public void onNext(Task task) { sink.success(task); } @Override public void onError(Throwable t) { logger.error("Error fetching task information for task ID = {}", taskId); apiErrors.incrementAndGet(); sink.error(t); } @Override public void onCompleted() { } })); } TitusClientImpl(JobManagementServiceStub jobManagementService,
                           JobManagementServiceFutureStub jobManagementServiceFutureStub,
                           Registry registry); @Override Mono<Task> getTask(String taskId); @Override Flux<JobOrTaskUpdate> getJobAndTaskUpdates(); @Override Mono<Job> getJobById(String jobId);  }
@Test public void getTaskById() { final CountDownLatch latch = new CountDownLatch(1); titusClient.getTask(taskOne.getId()).subscribe(task -> { assertThat(task.getId()).isEqualTo(taskOne.getId()); assertThat(task.getJobId()).isEqualTo(taskOne.getJobId()); latch.countDown(); }); try { latch.await(1, TimeUnit.SECONDS); } catch (InterruptedException e) { fail("getTaskById Timeout ", e); } }