@Test public void getReplica_existingReplica_returnsReplica() throws Exception { expect(partitionManager.getPartition(CLUSTER_ID, TOPIC_NAME, PARTITION_ID_1)) .andReturn(completedFuture(Optional.of(PARTITION_1))); replay(partitionManager); Optional<PartitionReplica> replica = replicaManager.getReplica(CLUSTER_ID, TOPIC_NAME, PARTITION_ID_1, REPLICA_1_1.getBrokerId()) .get(); assertEquals(REPLICA_1_1, replica.get()); }
@Override public CompletableFuture<Optional<PartitionReplica>> getReplica( String clusterId, String topicName, int partitionId, int brokerId) { return listReplicas(clusterId, topicName, partitionId) .thenApply(replicas -> findEntityByKey(replicas, PartitionReplica::getBrokerId, brokerId)); }
ReplicaManagerImpl implements ReplicaManager { @Override public CompletableFuture<Optional<PartitionReplica>> getReplica( String clusterId, String topicName, int partitionId, int brokerId) { return listReplicas(clusterId, topicName, partitionId) .thenApply(replicas -> findEntityByKey(replicas, PartitionReplica::getBrokerId, brokerId)); } }
ReplicaManagerImpl implements ReplicaManager { @Override public CompletableFuture<Optional<PartitionReplica>> getReplica( String clusterId, String topicName, int partitionId, int brokerId) { return listReplicas(clusterId, topicName, partitionId) .thenApply(replicas -> findEntityByKey(replicas, PartitionReplica::getBrokerId, brokerId)); } @Inject  ReplicaManagerImpl(
      Admin adminClient, BrokerManager brokerManager, PartitionManager partitionManager); }
ReplicaManagerImpl implements ReplicaManager { @Override public CompletableFuture<Optional<PartitionReplica>> getReplica( String clusterId, String topicName, int partitionId, int brokerId) { return listReplicas(clusterId, topicName, partitionId) .thenApply(replicas -> findEntityByKey(replicas, PartitionReplica::getBrokerId, brokerId)); } @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<Optional<PartitionReplica>> getReplica( String clusterId, String topicName, int partitionId, int brokerId) { return listReplicas(clusterId, topicName, partitionId) .thenApply(replicas -> findEntityByKey(replicas, PartitionReplica::getBrokerId, brokerId)); } @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);  }