@Test public void deleteTopic_nonExistingTopic_throwsUnknownTopicOrPartition() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER))); expect(adminClient.deleteTopics(singletonList(TOPIC_1.getName()))) .andReturn(deleteTopicsResult); expect(deleteTopicsResult.all()) .andReturn(failedFuture(new UnknownTopicOrPartitionException(""))); replay(clusterManager, adminClient, deleteTopicsResult); try { topicManager.deleteTopic(CLUSTER_ID, TOPIC_1.getName()).get(); fail(); } catch (ExecutionException e) { assertEquals(UnknownTopicOrPartitionException.class, e.getCause().getClass()); } verify(adminClient); }
@Override public CompletableFuture<Void> deleteTopic(String clusterId, String topicName) { requireNonNull(topicName); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.deleteTopics(singletonList(topicName)).all())); }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<Void> deleteTopic(String clusterId, String topicName) { requireNonNull(topicName); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.deleteTopics(singletonList(topicName)).all())); } }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<Void> deleteTopic(String clusterId, String topicName) { requireNonNull(topicName); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.deleteTopics(singletonList(topicName)).all())); } @Inject  TopicManagerImpl(Admin adminClient, ClusterManager clusterManager); }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<Void> deleteTopic(String clusterId, String topicName) { requireNonNull(topicName); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.deleteTopics(singletonList(topicName)).all())); } @Inject  TopicManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<Topic>> listTopics(String clusterId); @Override CompletableFuture<List<Topic>> listLocalTopics(); @Override CompletableFuture<Optional<Topic>> getTopic(String clusterId, String topicName); @Override CompletableFuture<Optional<Topic>> getLocalTopic(String topicName); @Override CompletableFuture<Void> createTopic(
      String clusterId,
      String topicName,
      int partitionsCount,
      short replicationFactor,
      Map<String, String> configs); @Override CompletableFuture<Void> deleteTopic(String clusterId, String topicName); }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<Void> deleteTopic(String clusterId, String topicName) { requireNonNull(topicName); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.deleteTopics(singletonList(topicName)).all())); } @Inject  TopicManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<Topic>> listTopics(String clusterId); @Override CompletableFuture<List<Topic>> listLocalTopics(); @Override CompletableFuture<Optional<Topic>> getTopic(String clusterId, String topicName); @Override CompletableFuture<Optional<Topic>> getLocalTopic(String topicName); @Override CompletableFuture<Void> createTopic(
      String clusterId,
      String topicName,
      int partitionsCount,
      short replicationFactor,
      Map<String, String> configs); @Override CompletableFuture<Void> deleteTopic(String clusterId, String topicName);  }