@Test public void searchByBrokerId_nonExistingBroker_throwsNotFound() throws Exception { expect(brokerManager.getBroker(CLUSTER_ID, BROKER_ID_1)) .andReturn(completedFuture(Optional.empty())); replay(brokerManager); try { replicaManager.searchReplicasByBrokerId(CLUSTER_ID, BROKER_ID_1).get(); fail(); } catch (ExecutionException e) { assertEquals(NotFoundException.class, e.getCause().getClass()); } }
@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);  }