@Test public void getPartition_nonExistingTopic_throwsNotFound() throws Exception { expect(topicManager.getTopic(CLUSTER_ID, TOPIC_NAME)) .andReturn(CompletableFuture.completedFuture(Optional.empty())); replay(topicManager); try { partitionManager.getPartition(CLUSTER_ID, TOPIC_NAME, PARTITION_1.getPartitionId()).get(); fail(); } catch (ExecutionException e) { assertEquals(NotFoundException.class, e.getCause().getClass()); } }
@Override public CompletableFuture<Optional<Partition>> getPartition( String clusterId, String topicName, int partitionId) { return topicManager.getTopic(clusterId, topicName) .thenApply(topic -> checkEntityExists(topic, "Topic %s cannot be found.", topic)) .thenApply(Topic::getPartitions) .thenApply( partitions -> findEntityByKey(partitions, Partition::getPartitionId, partitionId)) .thenApply( partition -> partition.map(Collections::singletonList).orElse(emptyList())) .thenCompose(this::withOffsets) .thenApply(partitions -> partitions.stream().findAny()); }
PartitionManagerImpl implements PartitionManager { @Override public CompletableFuture<Optional<Partition>> getPartition( String clusterId, String topicName, int partitionId) { return topicManager.getTopic(clusterId, topicName) .thenApply(topic -> checkEntityExists(topic, "Topic %s cannot be found.", topic)) .thenApply(Topic::getPartitions) .thenApply( partitions -> findEntityByKey(partitions, Partition::getPartitionId, partitionId)) .thenApply( partition -> partition.map(Collections::singletonList).orElse(emptyList())) .thenCompose(this::withOffsets) .thenApply(partitions -> partitions.stream().findAny()); } }
PartitionManagerImpl implements PartitionManager { @Override public CompletableFuture<Optional<Partition>> getPartition( String clusterId, String topicName, int partitionId) { return topicManager.getTopic(clusterId, topicName) .thenApply(topic -> checkEntityExists(topic, "Topic %s cannot be found.", topic)) .thenApply(Topic::getPartitions) .thenApply( partitions -> findEntityByKey(partitions, Partition::getPartitionId, partitionId)) .thenApply( partition -> partition.map(Collections::singletonList).orElse(emptyList())) .thenCompose(this::withOffsets) .thenApply(partitions -> partitions.stream().findAny()); } @Inject  PartitionManagerImpl(Admin adminClient, TopicManager topicManager); }
PartitionManagerImpl implements PartitionManager { @Override public CompletableFuture<Optional<Partition>> getPartition( String clusterId, String topicName, int partitionId) { return topicManager.getTopic(clusterId, topicName) .thenApply(topic -> checkEntityExists(topic, "Topic %s cannot be found.", topic)) .thenApply(Topic::getPartitions) .thenApply( partitions -> findEntityByKey(partitions, Partition::getPartitionId, partitionId)) .thenApply( partition -> partition.map(Collections::singletonList).orElse(emptyList())) .thenCompose(this::withOffsets) .thenApply(partitions -> partitions.stream().findAny()); } @Inject  PartitionManagerImpl(Admin adminClient, TopicManager topicManager); @Override CompletableFuture<List<Partition>> listPartitions(String clusterId, String topicName); @Override CompletableFuture<List<Partition>> listLocalPartitions(String topicName); @Override CompletableFuture<Optional<Partition>> getPartition(
      String clusterId, String topicName, int partitionId); @Override CompletableFuture<Optional<Partition>> getLocalPartition(
      String topicName, int partitionId); }
PartitionManagerImpl implements PartitionManager { @Override public CompletableFuture<Optional<Partition>> getPartition( String clusterId, String topicName, int partitionId) { return topicManager.getTopic(clusterId, topicName) .thenApply(topic -> checkEntityExists(topic, "Topic %s cannot be found.", topic)) .thenApply(Topic::getPartitions) .thenApply( partitions -> findEntityByKey(partitions, Partition::getPartitionId, partitionId)) .thenApply( partition -> partition.map(Collections::singletonList).orElse(emptyList())) .thenCompose(this::withOffsets) .thenApply(partitions -> partitions.stream().findAny()); } @Inject  PartitionManagerImpl(Admin adminClient, TopicManager topicManager); @Override CompletableFuture<List<Partition>> listPartitions(String clusterId, String topicName); @Override CompletableFuture<List<Partition>> listLocalPartitions(String topicName); @Override CompletableFuture<Optional<Partition>> getPartition(
      String clusterId, String topicName, int partitionId); @Override CompletableFuture<Optional<Partition>> getLocalPartition(
      String topicName, int partitionId);  }