@Test public void getConsumerAssignment_nonExistingConsumerAssignment_returnsEmpty() throws Exception { expect( consumerManager.getConsumer( CLUSTER_ID, CONSUMER.getConsumerGroupId(), CONSUMER.getConsumerId())) .andReturn(completedFuture(Optional.of(CONSUMER))); replay(consumerManager); Optional<ConsumerAssignment> assignment = consumerAssignmentManager.getConsumerAssignment( CLUSTER_ID, CONSUMER.getConsumerGroupId(), CONSUMER.getConsumerId(), "foobar", 100) .get(); assertFalse(assignment.isPresent()); }
@Override public CompletableFuture<Optional<ConsumerAssignment>> getConsumerAssignment( String clusterId, String consumerGroupId, String consumerId, String topicName, int partitionId ) { return listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> assignments.stream() .filter(assignment -> assignment.getTopicName().equals(topicName)) .filter(assignment -> assignment.getPartitionId() == partitionId) .findAny()); }
ConsumerAssignmentManagerImpl implements ConsumerAssignmentManager { @Override public CompletableFuture<Optional<ConsumerAssignment>> getConsumerAssignment( String clusterId, String consumerGroupId, String consumerId, String topicName, int partitionId ) { return listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> assignments.stream() .filter(assignment -> assignment.getTopicName().equals(topicName)) .filter(assignment -> assignment.getPartitionId() == partitionId) .findAny()); } }
ConsumerAssignmentManagerImpl implements ConsumerAssignmentManager { @Override public CompletableFuture<Optional<ConsumerAssignment>> getConsumerAssignment( String clusterId, String consumerGroupId, String consumerId, String topicName, int partitionId ) { return listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> assignments.stream() .filter(assignment -> assignment.getTopicName().equals(topicName)) .filter(assignment -> assignment.getPartitionId() == partitionId) .findAny()); } @Inject  ConsumerAssignmentManagerImpl(ConsumerManager consumerManager); }
ConsumerAssignmentManagerImpl implements ConsumerAssignmentManager { @Override public CompletableFuture<Optional<ConsumerAssignment>> getConsumerAssignment( String clusterId, String consumerGroupId, String consumerId, String topicName, int partitionId ) { return listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> assignments.stream() .filter(assignment -> assignment.getTopicName().equals(topicName)) .filter(assignment -> assignment.getPartitionId() == partitionId) .findAny()); } @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<Optional<ConsumerAssignment>> getConsumerAssignment( String clusterId, String consumerGroupId, String consumerId, String topicName, int partitionId ) { return listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> assignments.stream() .filter(assignment -> assignment.getTopicName().equals(topicName)) .filter(assignment -> assignment.getPartitionId() == partitionId) .findAny()); } @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
  );  }