@Test public void searchByBrokerId_existingBroker_returnsReplicas() throws Exception { HashMap<TopicPartition, ReplicaInfo> partitions = new HashMap<>(); partitions.put(new TopicPartition(TOPIC_NAME, PARTITION_ID_1), null); partitions.put(new TopicPartition(TOPIC_NAME, PARTITION_ID_2), null); expect(brokerManager.getBroker(CLUSTER_ID, BROKER_ID_1)) .andReturn(completedFuture(Optional.of(BROKER_1))); expect(adminClient.describeLogDirs(eq(singletonList(BROKER_ID_1)), anyObject())) .andReturn(describeLogDirsResult); expect(describeLogDirsResult.values()) .andReturn( singletonMap( BROKER_ID_1, KafkaFuture.completedFuture( singletonMap(TOPIC_NAME, new LogDirInfo(null, partitions))))); expect(partitionManager.getPartition(CLUSTER_ID, TOPIC_NAME, PARTITION_ID_1)) .andReturn(completedFuture(Optional.of(PARTITION_1))); expect(partitionManager.getPartition(CLUSTER_ID, TOPIC_NAME, PARTITION_ID_2)) .andReturn(completedFuture(Optional.of(PARTITION_2))); replay(adminClient, describeLogDirsResult, brokerManager, partitionManager); List<PartitionReplica> replicas = replicaManager.searchReplicasByBrokerId(CLUSTER_ID, BROKER_ID_1).get(); assertEquals(new HashSet<>(Arrays.asList(REPLICA_1_1, REPLICA_2_1)), new HashSet<>(replicas)); }
@Override public CompletableFuture<List<PartitionReplica>> searchReplicasByBrokerId( String clusterId, int brokerId) { return brokerManager.getBroker(clusterId, brokerId) .thenApply(broker -> checkEntityExists(broker, "Broker %d cannot be found.", brokerId)) .thenCompose( broker -> { DescribeLogDirsResult result = adminClient.describeLogDirs( singletonList(brokerId), new DescribeLogDirsOptions()); return KafkaFutures.toCompletableFuture(result.values().get(brokerId)); }) .thenCompose( logDirs -> CompletableFutures.allAsList( logDirs.values().stream() .flatMap(logDir -> logDir.replicaInfos.keySet().stream()) .map( partition -> getReplica( clusterId, partition.topic(), partition.partition(), brokerId)) .collect(Collectors.toList()))) .thenApply( replicas -> replicas.stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList())); }
ReplicaManagerImpl implements ReplicaManager { @Override public CompletableFuture<List<PartitionReplica>> searchReplicasByBrokerId( String clusterId, int brokerId) { return brokerManager.getBroker(clusterId, brokerId) .thenApply(broker -> checkEntityExists(broker, "Broker %d cannot be found.", brokerId)) .thenCompose( broker -> { DescribeLogDirsResult result = adminClient.describeLogDirs( singletonList(brokerId), new DescribeLogDirsOptions()); return KafkaFutures.toCompletableFuture(result.values().get(brokerId)); }) .thenCompose( logDirs -> CompletableFutures.allAsList( logDirs.values().stream() .flatMap(logDir -> logDir.replicaInfos.keySet().stream()) .map( partition -> getReplica( clusterId, partition.topic(), partition.partition(), brokerId)) .collect(Collectors.toList()))) .thenApply( replicas -> replicas.stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList())); } }
ReplicaManagerImpl implements ReplicaManager { @Override public CompletableFuture<List<PartitionReplica>> searchReplicasByBrokerId( String clusterId, int brokerId) { return brokerManager.getBroker(clusterId, brokerId) .thenApply(broker -> checkEntityExists(broker, "Broker %d cannot be found.", brokerId)) .thenCompose( broker -> { DescribeLogDirsResult result = adminClient.describeLogDirs( singletonList(brokerId), new DescribeLogDirsOptions()); return KafkaFutures.toCompletableFuture(result.values().get(brokerId)); }) .thenCompose( logDirs -> CompletableFutures.allAsList( logDirs.values().stream() .flatMap(logDir -> logDir.replicaInfos.keySet().stream()) .map( partition -> getReplica( clusterId, partition.topic(), partition.partition(), brokerId)) .collect(Collectors.toList()))) .thenApply( replicas -> replicas.stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList())); } @Inject  ReplicaManagerImpl(
      Admin adminClient, BrokerManager brokerManager, PartitionManager partitionManager); }
ReplicaManagerImpl implements ReplicaManager { @Override public CompletableFuture<List<PartitionReplica>> searchReplicasByBrokerId( String clusterId, int brokerId) { return brokerManager.getBroker(clusterId, brokerId) .thenApply(broker -> checkEntityExists(broker, "Broker %d cannot be found.", brokerId)) .thenCompose( broker -> { DescribeLogDirsResult result = adminClient.describeLogDirs( singletonList(brokerId), new DescribeLogDirsOptions()); return KafkaFutures.toCompletableFuture(result.values().get(brokerId)); }) .thenCompose( logDirs -> CompletableFutures.allAsList( logDirs.values().stream() .flatMap(logDir -> logDir.replicaInfos.keySet().stream()) .map( partition -> getReplica( clusterId, partition.topic(), partition.partition(), brokerId)) .collect(Collectors.toList()))) .thenApply( replicas -> replicas.stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList())); } @Inject  ReplicaManagerImpl(
      Admin adminClient, BrokerManager brokerManager, PartitionManager partitionManager); @Override CompletableFuture<List<PartitionReplica>> listReplicas(
      String clusterId, String topicName, int partitionId); @Override CompletableFuture<Optional<PartitionReplica>> getReplica(
      String clusterId, String topicName, int partitionId, int brokerId); @Override CompletableFuture<List<PartitionReplica>> searchReplicasByBrokerId(
      String clusterId, int brokerId); }
ReplicaManagerImpl implements ReplicaManager { @Override public CompletableFuture<List<PartitionReplica>> searchReplicasByBrokerId( String clusterId, int brokerId) { return brokerManager.getBroker(clusterId, brokerId) .thenApply(broker -> checkEntityExists(broker, "Broker %d cannot be found.", brokerId)) .thenCompose( broker -> { DescribeLogDirsResult result = adminClient.describeLogDirs( singletonList(brokerId), new DescribeLogDirsOptions()); return KafkaFutures.toCompletableFuture(result.values().get(brokerId)); }) .thenCompose( logDirs -> CompletableFutures.allAsList( logDirs.values().stream() .flatMap(logDir -> logDir.replicaInfos.keySet().stream()) .map( partition -> getReplica( clusterId, partition.topic(), partition.partition(), brokerId)) .collect(Collectors.toList()))) .thenApply( replicas -> replicas.stream() .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList())); } @Inject  ReplicaManagerImpl(
      Admin adminClient, BrokerManager brokerManager, PartitionManager partitionManager); @Override CompletableFuture<List<PartitionReplica>> listReplicas(
      String clusterId, String topicName, int partitionId); @Override CompletableFuture<Optional<PartitionReplica>> getReplica(
      String clusterId, String topicName, int partitionId, int brokerId); @Override CompletableFuture<List<PartitionReplica>> searchReplicasByBrokerId(
      String clusterId, int brokerId);  }