@Test public void listTopics_timeoutException_throwsTimeoutException() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER))); expect(adminClient.listTopics()).andReturn(listTopicsResult); expect(listTopicsResult.listings()).andReturn(failedFuture(new TimeoutException())); replay(clusterManager, adminClient, listTopicsResult); try { topicManager.listTopics(CLUSTER_ID).get(); fail(); } catch (ExecutionException e) { assertEquals(TimeoutException.class, e.getCause().getClass()); } }
@Override public CompletableFuture<List<Topic>> listTopics(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listTopics().listings())) .thenCompose( topicListings -> { if (topicListings == null) { return CompletableFuture.completedFuture(emptyList()); } return describeTopics( clusterId, topicListings.stream().map(TopicListing::name).collect(Collectors.toList())); }); }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<List<Topic>> listTopics(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listTopics().listings())) .thenCompose( topicListings -> { if (topicListings == null) { return CompletableFuture.completedFuture(emptyList()); } return describeTopics( clusterId, topicListings.stream().map(TopicListing::name).collect(Collectors.toList())); }); } }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<List<Topic>> listTopics(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listTopics().listings())) .thenCompose( topicListings -> { if (topicListings == null) { return CompletableFuture.completedFuture(emptyList()); } return describeTopics( clusterId, topicListings.stream().map(TopicListing::name).collect(Collectors.toList())); }); } @Inject  TopicManagerImpl(Admin adminClient, ClusterManager clusterManager); }
TopicManagerImpl implements TopicManager { @Override public CompletableFuture<List<Topic>> listTopics(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listTopics().listings())) .thenCompose( topicListings -> { if (topicListings == null) { return CompletableFuture.completedFuture(emptyList()); } return describeTopics( clusterId, topicListings.stream().map(TopicListing::name).collect(Collectors.toList())); }); } @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<List<Topic>> listTopics(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listTopics().listings())) .thenCompose( topicListings -> { if (topicListings == null) { return CompletableFuture.completedFuture(emptyList()); } return describeTopics( clusterId, topicListings.stream().map(TopicListing::name).collect(Collectors.toList())); }); } @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);  }