@Test public void listConsumerAssignments_returnsConsumerAssignments() throws Exception { expect( consumerManager.getConsumer( CLUSTER_ID, CONSUMER.getConsumerGroupId(), CONSUMER.getConsumerId())) .andReturn(completedFuture(Optional.of(CONSUMER))); replay(consumerManager); List<ConsumerAssignment> assignments = consumerAssignmentManager.listConsumerAssignments( CLUSTER_ID, CONSUMER.getConsumerGroupId(), CONSUMER.getConsumerId()).get(); assertEquals( Arrays.asList( ConsumerAssignment.builder() .setClusterId(CLUSTER_ID) .setConsumerGroupId(CONSUMER.getConsumerGroupId()) .setConsumerId(CONSUMER.getConsumerId()) .setTopicName("topic-1") .setPartitionId(1) .build(), ConsumerAssignment.builder() .setClusterId(CLUSTER_ID) .setConsumerGroupId(CONSUMER.getConsumerGroupId()) .setConsumerId(CONSUMER.getConsumerId()) .setTopicName("topic-2") .setPartitionId(2) .build(), ConsumerAssignment.builder() .setClusterId(CLUSTER_ID) .setConsumerGroupId(CONSUMER.getConsumerGroupId()) .setConsumerId(CONSUMER.getConsumerId()) .setTopicName("topic-3") .setPartitionId(3) .build()), assignments); }
@Override public CompletableFuture<List<ConsumerAssignment>> listConsumerAssignments( String clusterId, String consumerGroupId, String consumerId) { return consumerManager.getConsumer(clusterId, consumerGroupId, consumerId) .thenApply( consumer -> checkEntityExists(consumer, "Consumer %s does not exist.", consumerId)) .thenApply( consumer -> consumer.getAssignedPartitions() .stream() .map( partition -> ConsumerAssignment.builder() .setClusterId(clusterId) .setConsumerGroupId(consumerGroupId) .setConsumerId(consumerId) .setTopicName(partition.getTopicName()) .setPartitionId(partition.getPartitionId()) .build()) .collect(Collectors.toList())); }
ConsumerAssignmentManagerImpl implements ConsumerAssignmentManager { @Override public CompletableFuture<List<ConsumerAssignment>> listConsumerAssignments( String clusterId, String consumerGroupId, String consumerId) { return consumerManager.getConsumer(clusterId, consumerGroupId, consumerId) .thenApply( consumer -> checkEntityExists(consumer, "Consumer %s does not exist.", consumerId)) .thenApply( consumer -> consumer.getAssignedPartitions() .stream() .map( partition -> ConsumerAssignment.builder() .setClusterId(clusterId) .setConsumerGroupId(consumerGroupId) .setConsumerId(consumerId) .setTopicName(partition.getTopicName()) .setPartitionId(partition.getPartitionId()) .build()) .collect(Collectors.toList())); } }
ConsumerAssignmentManagerImpl implements ConsumerAssignmentManager { @Override public CompletableFuture<List<ConsumerAssignment>> listConsumerAssignments( String clusterId, String consumerGroupId, String consumerId) { return consumerManager.getConsumer(clusterId, consumerGroupId, consumerId) .thenApply( consumer -> checkEntityExists(consumer, "Consumer %s does not exist.", consumerId)) .thenApply( consumer -> consumer.getAssignedPartitions() .stream() .map( partition -> ConsumerAssignment.builder() .setClusterId(clusterId) .setConsumerGroupId(consumerGroupId) .setConsumerId(consumerId) .setTopicName(partition.getTopicName()) .setPartitionId(partition.getPartitionId()) .build()) .collect(Collectors.toList())); } @Inject  ConsumerAssignmentManagerImpl(ConsumerManager consumerManager); }
ConsumerAssignmentManagerImpl implements ConsumerAssignmentManager { @Override public CompletableFuture<List<ConsumerAssignment>> listConsumerAssignments( String clusterId, String consumerGroupId, String consumerId) { return consumerManager.getConsumer(clusterId, consumerGroupId, consumerId) .thenApply( consumer -> checkEntityExists(consumer, "Consumer %s does not exist.", consumerId)) .thenApply( consumer -> consumer.getAssignedPartitions() .stream() .map( partition -> ConsumerAssignment.builder() .setClusterId(clusterId) .setConsumerGroupId(consumerGroupId) .setConsumerId(consumerId) .setTopicName(partition.getTopicName()) .setPartitionId(partition.getPartitionId()) .build()) .collect(Collectors.toList())); } @Inject  ConsumerAssignmentManagerImpl(ConsumerManager consumerManager); @Override CompletableFuture<List<ConsumerAssignment>> listConsumerAssignments(
      String clusterId, String consumerGroupId, String consumerId); @Override CompletableFuture<Optional<ConsumerAssignment>> getConsumerAssignment(
      String clusterId,
      String consumerGroupId,
      String consumerId,
      String topicName,
      int partitionId
  ); }
ConsumerAssignmentManagerImpl implements ConsumerAssignmentManager { @Override public CompletableFuture<List<ConsumerAssignment>> listConsumerAssignments( String clusterId, String consumerGroupId, String consumerId) { return consumerManager.getConsumer(clusterId, consumerGroupId, consumerId) .thenApply( consumer -> checkEntityExists(consumer, "Consumer %s does not exist.", consumerId)) .thenApply( consumer -> consumer.getAssignedPartitions() .stream() .map( partition -> ConsumerAssignment.builder() .setClusterId(clusterId) .setConsumerGroupId(consumerGroupId) .setConsumerId(consumerId) .setTopicName(partition.getTopicName()) .setPartitionId(partition.getPartitionId()) .build()) .collect(Collectors.toList())); } @Inject  ConsumerAssignmentManagerImpl(ConsumerManager consumerManager); @Override CompletableFuture<List<ConsumerAssignment>> listConsumerAssignments(
      String clusterId, String consumerGroupId, String consumerId); @Override CompletableFuture<Optional<ConsumerAssignment>> getConsumerAssignment(
      String clusterId,
      String consumerGroupId,
      String consumerId,
      String topicName,
      int partitionId
  );  }