@Test public void createTopic_nonExistingCluster_throwsNotFound() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.empty())); replay(clusterManager); try { topicManager.createTopic( CLUSTER_ID, TOPIC_1.getName(), TOPIC_1.getPartitions().size(), TOPIC_1.getReplicationFactor(), singletonMap("cleanup.policy", "compact")).get(); fail(); } catch (ExecutionException e) { assertEquals(NotFoundException.class, e.getCause().getClass()); } }
@Override public CompletableFuture<Void> createTopic( String clusterId, String topicName, int partitionsCount, short replicationFactor, Map<String, String> configs) { requireNonNull(topicName); NewTopic createTopicRequest = new NewTopic(topicName, partitionsCount, replicationFactor).configs(configs); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.createTopics(singletonList(createTopicRequest)).all())); }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<Void> createTopic( String clusterId, String topicName, int partitionsCount, short replicationFactor, Map<String, String> configs) { requireNonNull(topicName); NewTopic createTopicRequest = new NewTopic(topicName, partitionsCount, replicationFactor).configs(configs); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.createTopics(singletonList(createTopicRequest)).all())); } }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<Void> createTopic( String clusterId, String topicName, int partitionsCount, short replicationFactor, Map<String, String> configs) { requireNonNull(topicName); NewTopic createTopicRequest = new NewTopic(topicName, partitionsCount, replicationFactor).configs(configs); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.createTopics(singletonList(createTopicRequest)).all())); } @Inject  TopicManagerImpl(Admin adminClient, ClusterManager clusterManager); }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<Void> createTopic( String clusterId, String topicName, int partitionsCount, short replicationFactor, Map<String, String> configs) { requireNonNull(topicName); NewTopic createTopicRequest = new NewTopic(topicName, partitionsCount, replicationFactor).configs(configs); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.createTopics(singletonList(createTopicRequest)).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> createTopic( String clusterId, String topicName, int partitionsCount, short replicationFactor, Map<String, String> configs) { requireNonNull(topicName); NewTopic createTopicRequest = new NewTopic(topicName, partitionsCount, replicationFactor).configs(configs); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture( adminClient.createTopics(singletonList(createTopicRequest)).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);  }