@Test public void searchReassignmentsByTopic_existingCluster_returnsReassignments() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER))); expect(adminClient.listPartitionReassignments()).andReturn(listPartitionReassignmentsResult); expect(listPartitionReassignmentsResult.reassignments()) .andReturn(KafkaFuture.completedFuture(REASSIGNMENT_MAP)); replay(clusterManager, adminClient, listPartitionReassignmentsResult); List<Reassignment> reassignments = reassignmentManager.searchReassignmentsByTopicName(CLUSTER_ID, TOPIC_1).get(); assertEquals(Arrays.asList(REASSIGNMENT_1, REASSIGNMENT_2, REASSIGNMENT_3), reassignments); }
@Override public CompletableFuture<List<Reassignment>> searchReassignmentsByTopicName( String clusterId, String topicName) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .sorted( Comparator.comparing(Reassignment::getTopicName) .thenComparing(Reassignment::getPartitionId)) .collect(Collectors.toList())); }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<List<Reassignment>> searchReassignmentsByTopicName( String clusterId, String topicName) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .sorted( Comparator.comparing(Reassignment::getTopicName) .thenComparing(Reassignment::getPartitionId)) .collect(Collectors.toList())); } }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<List<Reassignment>> searchReassignmentsByTopicName( String clusterId, String topicName) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .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>> searchReassignmentsByTopicName( String clusterId, String topicName) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .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>> searchReassignmentsByTopicName( String clusterId, String topicName) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .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);  }