@Test public void getConsumerGroup_returnsConsumerGroup() 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( singletonMap( CONSUMER_GROUPS[0].getConsumerGroupId(), consumerGroupDescriptions[0]))); expect(consumerGroupDescriptions[0].groupId()) .andStubReturn(CONSUMER_GROUPS[0].getConsumerGroupId()); expect(consumerGroupDescriptions[0].isSimpleConsumerGroup()) .andStubReturn(CONSUMER_GROUPS[0].isSimple()); expect(consumerGroupDescriptions[0].partitionAssignor()) .andStubReturn(CONSUMER_GROUPS[0].getPartitionAssignor()); expect(consumerGroupDescriptions[0].state()) .andStubReturn(CONSUMER_GROUPS[0].getState().toConsumerGroupState()); expect(consumerGroupDescriptions[0].coordinator()) .andStubReturn(CONSUMER_GROUPS[0].getCoordinator().toNode()); expect(consumerGroupDescriptions[0].members()) .andStubReturn(Arrays.asList(memberDescriptions[0])); for (int j = 0; j < CONSUMER_GROUPS[0].getConsumers().size(); j++) { expect(memberDescriptions[0][j].consumerId()) .andStubReturn(CONSUMERS[0][j].getConsumerId()); expect(memberDescriptions[0][j].groupInstanceId()) .andStubReturn(CONSUMERS[0][j].getInstanceId()); expect(memberDescriptions[0][j].clientId()).andStubReturn(CONSUMERS[0][j].getClientId()); expect(memberDescriptions[0][j].host()).andStubReturn(CONSUMERS[0][j].getHost()); expect(memberDescriptions[0][j].assignment()).andStubReturn(memberAssignments[0][j]); expect(memberAssignments[0][j].topicPartitions()) .andStubReturn( CONSUMERS[0][j].getAssignedPartitions() .stream() .map(Partition::toTopicPartition) .collect(Collectors.toSet())); replay(memberDescriptions[0][j], memberAssignments[0][j]); } replay( clusterManager, adminClient, listConsumerGroupsResult, describeConsumerGroupsResult, consumerGroupDescriptions[0]); ConsumerGroup consumerGroup = consumerGroupManager.getConsumerGroup(CLUSTER_ID, CONSUMER_GROUPS[0].getConsumerGroupId()) .get() .get(); assertEquals(CONSUMER_GROUPS[0], consumerGroup); }
@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);  }