@Test public void getConsumerGroup_nonExistingCluster_throwsNotFound() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.empty())); replay(clusterManager); try { consumerGroupManager.getConsumerGroup(CLUSTER_ID, CONSUMER_GROUPS[0].getConsumerGroupId()) .get(); fail(); } catch (ExecutionException e) { assertEquals(NotFoundException.class, e.getCause().getClass()); } }
@Override public CompletableFuture<Optional<ConsumerGroup>> getConsumerGroup( String clusterId, String consumerGroupId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose(cluster -> getConsumerGroups(clusterId, singletonList(consumerGroupId))) .thenApply(consumerGroups -> consumerGroups.stream().findAny()); }
ConsumerGroupManagerImpl implements ConsumerGroupManager { @Override public CompletableFuture<Optional<ConsumerGroup>> getConsumerGroup( String clusterId, String consumerGroupId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose(cluster -> getConsumerGroups(clusterId, singletonList(consumerGroupId))) .thenApply(consumerGroups -> consumerGroups.stream().findAny()); } }
ConsumerGroupManagerImpl implements ConsumerGroupManager { @Override public CompletableFuture<Optional<ConsumerGroup>> getConsumerGroup( String clusterId, String consumerGroupId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose(cluster -> getConsumerGroups(clusterId, singletonList(consumerGroupId))) .thenApply(consumerGroups -> consumerGroups.stream().findAny()); } @Inject  ConsumerGroupManagerImpl(Admin adminClient, ClusterManager clusterManager); }
ConsumerGroupManagerImpl implements ConsumerGroupManager { @Override public CompletableFuture<Optional<ConsumerGroup>> getConsumerGroup( String clusterId, String consumerGroupId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose(cluster -> getConsumerGroups(clusterId, singletonList(consumerGroupId))) .thenApply(consumerGroups -> consumerGroups.stream().findAny()); } @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<Optional<ConsumerGroup>> getConsumerGroup( String clusterId, String consumerGroupId) { return clusterManager.getCluster(clusterId) .thenApply( cluster -> checkEntityExists(cluster, "Cluster %s could not be found.", clusterId)) .thenCompose(cluster -> getConsumerGroups(clusterId, singletonList(consumerGroupId))) .thenApply(consumerGroups -> consumerGroups.stream().findAny()); } @Inject  ConsumerGroupManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<ConsumerGroup>> listConsumerGroups(String clusterId); @Override CompletableFuture<Optional<ConsumerGroup>> getConsumerGroup(
      String clusterId, String consumerGroupId);  }