ClientClusterManagementService implements ClusterManagementService { @Override public <A extends ClusterManagementOperation<V>, V extends OperationResult> CompletableFuture<ClusterManagementOperationResult<A, V>> getFuture( A opType, String opId) { AtomicReference<CompletableFuture<ClusterManagementOperationResult<A, V>>> futureAtomicReference = new AtomicReference<>(); futureAtomicReference.set(CompletableFuture.supplyAsync(() -> { while (futureAtomicReference.get() == null || !futureAtomicReference.get().isCancelled()) { ClusterManagementOperationResult<A, V> result = this.get(opType, opId); if (result.getOperationEnd() != null) { return result; } try { Thread.sleep(1000L); } catch (InterruptedException e) { throw new ClusterManagementException(result, e); } } return null; })); return futureAtomicReference.get(); } ClientClusterManagementService(ClusterManagementServiceTransport transport); @Override ClusterManagementRealizationResult create(T config); @Override ClusterManagementRealizationResult delete(
      T config); @Override ClusterManagementRealizationResult update(
      T config); @Override ClusterManagementListResult<T, R> list(
      T config); @Override ClusterManagementGetResult<T, R> get(
      T config); @Override ClusterManagementOperationResult<A, V> start(
      A op); @Override ClusterManagementOperationResult<A, V> get(
      A opType, String opId); @Override CompletableFuture<ClusterManagementOperationResult<A, V>> getFuture(
      A opType, String opId); @Override ClusterManagementListOperationsResult<A, V> list(
      A opType); @Override boolean isConnected(); @Override void close();  }
@Test public void getOperationCallsSubmitMessageAndReturnsFuture() { String opId = "opId"; RebalanceOperation opType = new RebalanceOperation(); doReturn(successOperationResult).when(serviceTransport) .submitMessageForGetOperation(same(opType), same(opId)); CompletableFuture<ClusterManagementOperationResult<RebalanceOperation, RebalanceResult>> future = service.getFuture(opType, opId); await().untilAsserted( () -> verify(serviceTransport, atLeastOnce()).submitMessageForGetOperation(same(opType), same(opId))); assertThat(future.isDone()).isFalse(); future.cancel(true); }