@Test public void listConsumerGroups_returnsConsumerGroups() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER))); expect(adminClient.listConsumerGroups()).andReturn(listConsumerGroupsResult); expect(listConsumerGroupsResult.all()) .andReturn(KafkaFuture.completedFuture(Arrays.asList(consumerGroupListings))); for (int i = 0; i < CONSUMER_GROUPS.length; i++) { expect(consumerGroupListings[i].groupId()).andReturn(CONSUMER_GROUPS[i].getConsumerGroupId()); replay(consumerGroupListings[i]); } expect( adminClient.describeConsumerGroups( Arrays.stream(CONSUMER_GROUPS) .map(ConsumerGroup::getConsumerGroupId) .collect(Collectors.toList()))) .andReturn(describeConsumerGroupsResult); expect(describeConsumerGroupsResult.all()) .andReturn( KafkaFuture.completedFuture( IntStream.range(0, CONSUMER_GROUPS.length) .boxed() .collect( Collectors.toMap( i -> CONSUMER_GROUPS[i].getConsumerGroupId(), i -> consumerGroupDescriptions[i])))); for (int i = 0; i < CONSUMER_GROUPS.length; i++) { expect(consumerGroupDescriptions[i].groupId()) .andStubReturn(CONSUMER_GROUPS[i].getConsumerGroupId()); expect(consumerGroupDescriptions[i].isSimpleConsumerGroup()) .andStubReturn(CONSUMER_GROUPS[i].isSimple()); expect(consumerGroupDescriptions[i].partitionAssignor()) .andStubReturn(CONSUMER_GROUPS[i].getPartitionAssignor()); expect(consumerGroupDescriptions[i].state()) .andStubReturn(CONSUMER_GROUPS[i].getState().toConsumerGroupState()); expect(consumerGroupDescriptions[i].coordinator()) .andStubReturn(CONSUMER_GROUPS[i].getCoordinator().toNode()); expect(consumerGroupDescriptions[i].members()) .andStubReturn(Arrays.asList(memberDescriptions[i])); replay(consumerGroupDescriptions[i]); } for (int i = 0; i < CONSUMER_GROUPS.length; i++) { for (int j = 0; j < CONSUMER_GROUPS[i].getConsumers().size(); j++) { expect(memberDescriptions[i][j].consumerId()) .andStubReturn(CONSUMERS[i][j].getConsumerId()); expect(memberDescriptions[i][j].groupInstanceId()) .andStubReturn(CONSUMERS[i][j].getInstanceId()); expect(memberDescriptions[i][j].clientId()).andStubReturn(CONSUMERS[i][j].getClientId()); expect(memberDescriptions[i][j].host()).andStubReturn(CONSUMERS[i][j].getHost()); expect(memberDescriptions[i][j].assignment()).andStubReturn(memberAssignments[i][j]); expect(memberAssignments[i][j].topicPartitions()) .andStubReturn( CONSUMERS[i][j].getAssignedPartitions() .stream() .map(Partition::toTopicPartition) .collect(Collectors.toSet())); replay(memberDescriptions[i][j], memberAssignments[i][j]); } } replay(clusterManager, adminClient, listConsumerGroupsResult, describeConsumerGroupsResult); List<ConsumerGroup> consumerGroups = consumerGroupManager.listConsumerGroups(CLUSTER_ID).get(); assertEquals(Arrays.asList(CONSUMER_GROUPS), consumerGroups); }
@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);  }