@Test public void listConsumerGroups_nonExistentCluster_throwsNotFound() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.empty())); replay(clusterManager); try { consumerGroupManager.listConsumerGroups(CLUSTER_ID).get(); fail(); } catch (ExecutionException e) { assertEquals(NotFoundException.class, e.getCause().getClass()); } }
@Override public CompletableFuture<List<ConsumerGroup>> listConsumerGroups(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listConsumerGroups().all())) .thenCompose( listings -> getConsumerGroups( clusterId, listings.stream() .map(ConsumerGroupListing::groupId) .collect(Collectors.toList()))); }
ConsumerGroupManagerImpl implements ConsumerGroupManager { @Override public CompletableFuture<List<ConsumerGroup>> listConsumerGroups(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listConsumerGroups().all())) .thenCompose( listings -> getConsumerGroups( clusterId, listings.stream() .map(ConsumerGroupListing::groupId) .collect(Collectors.toList()))); } }
ConsumerGroupManagerImpl implements ConsumerGroupManager { @Override public CompletableFuture<List<ConsumerGroup>> listConsumerGroups(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listConsumerGroups().all())) .thenCompose( listings -> getConsumerGroups( clusterId, listings.stream() .map(ConsumerGroupListing::groupId) .collect(Collectors.toList()))); } @Inject  ConsumerGroupManagerImpl(Admin adminClient, ClusterManager clusterManager); }
ConsumerGroupManagerImpl implements ConsumerGroupManager { @Override public CompletableFuture<List<ConsumerGroup>> listConsumerGroups(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listConsumerGroups().all())) .thenCompose( listings -> getConsumerGroups( clusterId, listings.stream() .map(ConsumerGroupListing::groupId) .collect(Collectors.toList()))); } @Inject  ConsumerGroupManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<ConsumerGroup>> listConsumerGroups(String clusterId); @Override CompletableFuture<Optional<ConsumerGroup>> getConsumerGroup(
      String clusterId, String consumerGroupId); }
ConsumerGroupManagerImpl implements ConsumerGroupManager { @Override public CompletableFuture<List<ConsumerGroup>> listConsumerGroups(String clusterId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose( cluster -> KafkaFutures.toCompletableFuture(adminClient.listConsumerGroups().all())) .thenCompose( listings -> getConsumerGroups( clusterId, listings.stream() .map(ConsumerGroupListing::groupId) .collect(Collectors.toList()))); } @Inject  ConsumerGroupManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<ConsumerGroup>> listConsumerGroups(String clusterId); @Override CompletableFuture<Optional<ConsumerGroup>> getConsumerGroup(
      String clusterId, String consumerGroupId);  }