@Test public void listAllReassignments_timeoutException_throwsTimeoutException() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER))); expect(adminClient.listPartitionReassignments()).andReturn(listPartitionReassignmentsResult); expect(listPartitionReassignmentsResult.reassignments()) .andReturn(failedFuture(new TimeoutException())); replay(clusterManager, adminClient, listPartitionReassignmentsResult); try { reassignmentManager.listReassignments(CLUSTER_ID).get(); fail(); } catch (ExecutionException e) { assertEquals(TimeoutException.class, e.getCause().getClass()); } }
@Override public CompletableFuture<List<Reassignment>> listReassignments( String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures .toCompletableFuture(adminClient.listPartitionReassignments().reassignments())) .thenApply( reassignments -> { if (reassignments == null) { return emptyList(); } return reassignments.entrySet().stream() .map(reassignment -> toReassignment(clusterId, reassignment)) .sorted( Comparator.comparing(Reassignment::getTopicName) .thenComparing(Reassignment::getPartitionId)) .collect(Collectors.toList()); }); }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<List<Reassignment>> listReassignments( String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures .toCompletableFuture(adminClient.listPartitionReassignments().reassignments())) .thenApply( reassignments -> { if (reassignments == null) { return emptyList(); } return reassignments.entrySet().stream() .map(reassignment -> toReassignment(clusterId, reassignment)) .sorted( Comparator.comparing(Reassignment::getTopicName) .thenComparing(Reassignment::getPartitionId)) .collect(Collectors.toList()); }); } }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<List<Reassignment>> listReassignments( String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures .toCompletableFuture(adminClient.listPartitionReassignments().reassignments())) .thenApply( reassignments -> { if (reassignments == null) { return emptyList(); } return reassignments.entrySet().stream() .map(reassignment -> toReassignment(clusterId, reassignment)) .sorted( Comparator.comparing(Reassignment::getTopicName) .thenComparing(Reassignment::getPartitionId)) .collect(Collectors.toList()); }); } @Inject  ReassignmentManagerImpl(Admin adminClient,
      ClusterManager clusterManager); }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<List<Reassignment>> listReassignments( String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures .toCompletableFuture(adminClient.listPartitionReassignments().reassignments())) .thenApply( reassignments -> { if (reassignments == null) { return emptyList(); } return reassignments.entrySet().stream() .map(reassignment -> toReassignment(clusterId, reassignment)) .sorted( Comparator.comparing(Reassignment::getTopicName) .thenComparing(Reassignment::getPartitionId)) .collect(Collectors.toList()); }); } @Inject  ReassignmentManagerImpl(Admin adminClient,
      ClusterManager clusterManager); @Override CompletableFuture<List<Reassignment>> listReassignments(
      String clusterId); @Override CompletableFuture<List<Reassignment>> searchReassignmentsByTopicName(
      String clusterId, String topicName); @Override CompletableFuture<Optional<Reassignment>> getReassignment(
      String clusterId, String topicName, Integer partitionId); }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<List<Reassignment>> listReassignments( String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures .toCompletableFuture(adminClient.listPartitionReassignments().reassignments())) .thenApply( reassignments -> { if (reassignments == null) { return emptyList(); } return reassignments.entrySet().stream() .map(reassignment -> toReassignment(clusterId, reassignment)) .sorted( Comparator.comparing(Reassignment::getTopicName) .thenComparing(Reassignment::getPartitionId)) .collect(Collectors.toList()); }); } @Inject  ReassignmentManagerImpl(Admin adminClient,
      ClusterManager clusterManager); @Override CompletableFuture<List<Reassignment>> listReassignments(
      String clusterId); @Override CompletableFuture<List<Reassignment>> searchReassignmentsByTopicName(
      String clusterId, String topicName); @Override CompletableFuture<Optional<Reassignment>> getReassignment(
      String clusterId, String topicName, Integer partitionId);  }