@Test public void getConsumerGroup_nonExistingConsumerGroup_returnsEmpty() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER))); expect( adminClient.describeConsumerGroups( singletonList(CONSUMER_GROUPS[0].getConsumerGroupId()))) .andReturn(describeConsumerGroupsResult); expect(describeConsumerGroupsResult.all()).andReturn(KafkaFuture.completedFuture(emptyMap())); replay(clusterManager, adminClient, listConsumerGroupsResult, describeConsumerGroupsResult); Optional<ConsumerGroup> consumerGroup = consumerGroupManager.getConsumerGroup(CLUSTER_ID, CONSUMER_GROUPS[0].getConsumerGroupId()) .get(); assertFalse(consumerGroup.isPresent()); }
@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);  }