@Test public void getReassignment_nonExistingTopic_returnsEmpty() 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); Optional<Reassignment> reassignment = reassignmentManager.getReassignment(CLUSTER_ID, "foobar", 3).get(); assertFalse(reassignment.isPresent()); }
@Override public CompletableFuture<Optional<Reassignment>> getReassignment( String clusterId, String topicName, Integer partitionId) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .filter(reassignment -> reassignment.getPartitionId() == partitionId) .findAny()); }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<Optional<Reassignment>> getReassignment( String clusterId, String topicName, Integer partitionId) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .filter(reassignment -> reassignment.getPartitionId() == partitionId) .findAny()); } }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<Optional<Reassignment>> getReassignment( String clusterId, String topicName, Integer partitionId) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .filter(reassignment -> reassignment.getPartitionId() == partitionId) .findAny()); } @Inject  ReassignmentManagerImpl(Admin adminClient,
      ClusterManager clusterManager); }
ReassignmentManagerImpl implements ReassignmentManager { @Override public CompletableFuture<Optional<Reassignment>> getReassignment( String clusterId, String topicName, Integer partitionId) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .filter(reassignment -> reassignment.getPartitionId() == partitionId) .findAny()); } @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<Optional<Reassignment>> getReassignment( String clusterId, String topicName, Integer partitionId) { return listReassignments(clusterId) .thenApply(reassignments -> reassignments.stream() .filter(reassignment -> reassignment.getTopicName().equals(topicName)) .filter(reassignment -> reassignment.getPartitionId() == partitionId) .findAny()); } @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);  }