@Test public void listConsumerAssignments_nonExistentConsumer_throwsNotFound() throws Exception { expect( consumerManager.getConsumer( CLUSTER_ID, CONSUMER.getConsumerGroupId(), CONSUMER.getConsumerId())) .andReturn(completedFuture(Optional.empty())); replay(consumerManager); try { consumerAssignmentManager.listConsumerAssignments( CLUSTER_ID, CONSUMER.getConsumerGroupId(), CONSUMER.getConsumerId()).get(); fail(); } catch (ExecutionException e) { assertEquals(NotFoundException.class, e.getCause().getClass()); } }
@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
  );  }