@Test public void listConsumers_returnsConsumers() { expect( consumerAssignmentManager.listConsumerAssignments( CLUSTER_ID, CONSUMER_GROUP_ID, CONSUMER_ID)) .andReturn( completedFuture( Arrays.asList( CONSUMER_ASSIGNMENT_1, CONSUMER_ASSIGNMENT_2, CONSUMER_ASSIGNMENT_3))); replay(consumerAssignmentManager); FakeAsyncResponse response = new FakeAsyncResponse(); consumerAssignmentsResource.listConsumerAssignments( response, CLUSTER_ID, CONSUMER_GROUP_ID, CONSUMER_ID); ListConsumerAssignmentsResponse expected = ListConsumerAssignmentsResponse.create( ConsumerAssignmentDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( "/v3/clusters/cluster-1/consumer-groups/consumer-group-1" + "/consumers/consumer-1/assignments") .build()) .setData( Arrays.asList( ConsumerAssignmentData.fromConsumerAssignment(CONSUMER_ASSIGNMENT_1) .setMetadata( Resource.Metadata.builder() .setSelf( "/v3/clusters/cluster-1/consumer-groups/consumer-group-1" + "/consumers/consumer-1" + "/assignments/topic-1/partitions/1") .setResourceName( "crn: + "/consumer=consumer-1" + "/assignment=topic-1/partition=1") .build()) .setPartition( Relationship.create( "/v3/clusters/cluster-1/topics/topic-1/partitions/1")) .build(), ConsumerAssignmentData.fromConsumerAssignment(CONSUMER_ASSIGNMENT_2) .setMetadata( Resource.Metadata.builder() .setSelf( "/v3/clusters/cluster-1/consumer-groups/consumer-group-1" + "/consumers/consumer-1" + "/assignments/topic-2/partitions/2") .setResourceName( "crn: + "/consumer=consumer-1" + "/assignment=topic-2/partition=2") .build()) .setPartition( Relationship.create( "/v3/clusters/cluster-1/topics/topic-2/partitions/2")) .build(), ConsumerAssignmentData.fromConsumerAssignment(CONSUMER_ASSIGNMENT_3) .setMetadata( Resource.Metadata.builder() .setSelf( "/v3/clusters/cluster-1/consumer-groups/consumer-group-1" + "/consumers/consumer-1" + "/assignments/topic-3/partitions/3") .setResourceName( "crn: + "/consumer=consumer-1" + "/assignment=topic-3/partition=3") .build()) .setPartition( Relationship.create( "/v3/clusters/cluster-1/topics/topic-3/partitions/3")) .build())) .build()); assertEquals(expected, response.getValue()); }
@GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.list") public void listConsumerAssignments( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("consumerGroupId") String consumerGroupId, @PathParam("consumerId") String consumerId ) { CompletableFuture<ListConsumerAssignmentsResponse> response = consumerAssignmentManager.get() .listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> ListConsumerAssignmentsResponse.create( ConsumerAssignmentDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "consumer-groups", consumerGroupId, "consumers", consumerId, "assignments")) .build()) .setData( assignments.stream() .map(this::toConsumerAssignmentData) .sorted( Comparator.comparing(ConsumerAssignmentData::getTopicName) .thenComparing(ConsumerAssignmentData::getPartitionId)) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); }
ConsumerAssignmentsResource { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.list") public void listConsumerAssignments( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("consumerGroupId") String consumerGroupId, @PathParam("consumerId") String consumerId ) { CompletableFuture<ListConsumerAssignmentsResponse> response = consumerAssignmentManager.get() .listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> ListConsumerAssignmentsResponse.create( ConsumerAssignmentDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "consumer-groups", consumerGroupId, "consumers", consumerId, "assignments")) .build()) .setData( assignments.stream() .map(this::toConsumerAssignmentData) .sorted( Comparator.comparing(ConsumerAssignmentData::getTopicName) .thenComparing(ConsumerAssignmentData::getPartitionId)) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } }
ConsumerAssignmentsResource { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.list") public void listConsumerAssignments( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("consumerGroupId") String consumerGroupId, @PathParam("consumerId") String consumerId ) { CompletableFuture<ListConsumerAssignmentsResponse> response = consumerAssignmentManager.get() .listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> ListConsumerAssignmentsResponse.create( ConsumerAssignmentDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "consumer-groups", consumerGroupId, "consumers", consumerId, "assignments")) .build()) .setData( assignments.stream() .map(this::toConsumerAssignmentData) .sorted( Comparator.comparing(ConsumerAssignmentData::getTopicName) .thenComparing(ConsumerAssignmentData::getPartitionId)) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  ConsumerAssignmentsResource(
      Provider<ConsumerAssignmentManager> consumerAssignmentManager,
      CrnFactory crnFactory,
      UrlFactory urlFactory
  ); }
ConsumerAssignmentsResource { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.list") public void listConsumerAssignments( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("consumerGroupId") String consumerGroupId, @PathParam("consumerId") String consumerId ) { CompletableFuture<ListConsumerAssignmentsResponse> response = consumerAssignmentManager.get() .listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> ListConsumerAssignmentsResponse.create( ConsumerAssignmentDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "consumer-groups", consumerGroupId, "consumers", consumerId, "assignments")) .build()) .setData( assignments.stream() .map(this::toConsumerAssignmentData) .sorted( Comparator.comparing(ConsumerAssignmentData::getTopicName) .thenComparing(ConsumerAssignmentData::getPartitionId)) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  ConsumerAssignmentsResource(
      Provider<ConsumerAssignmentManager> consumerAssignmentManager,
      CrnFactory crnFactory,
      UrlFactory urlFactory
  ); @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.list") void listConsumerAssignments(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("consumerGroupId") String consumerGroupId,
      @PathParam("consumerId") String consumerId
  ); @GET @Path("/{topicName}/partitions/{partitionId}") @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.get") void getConsumerAssignment(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("consumerGroupId") String consumerGroupId,
      @PathParam("consumerId") String consumerId,
      @PathParam("topicName") String topicName,
      @PathParam("partitionId") int partitionId
  ); }
ConsumerAssignmentsResource { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.list") public void listConsumerAssignments( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("consumerGroupId") String consumerGroupId, @PathParam("consumerId") String consumerId ) { CompletableFuture<ListConsumerAssignmentsResponse> response = consumerAssignmentManager.get() .listConsumerAssignments(clusterId, consumerGroupId, consumerId) .thenApply( assignments -> ListConsumerAssignmentsResponse.create( ConsumerAssignmentDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "consumer-groups", consumerGroupId, "consumers", consumerId, "assignments")) .build()) .setData( assignments.stream() .map(this::toConsumerAssignmentData) .sorted( Comparator.comparing(ConsumerAssignmentData::getTopicName) .thenComparing(ConsumerAssignmentData::getPartitionId)) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  ConsumerAssignmentsResource(
      Provider<ConsumerAssignmentManager> consumerAssignmentManager,
      CrnFactory crnFactory,
      UrlFactory urlFactory
  ); @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.list") void listConsumerAssignments(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("consumerGroupId") String consumerGroupId,
      @PathParam("consumerId") String consumerId
  ); @GET @Path("/{topicName}/partitions/{partitionId}") @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.consumer-assignments.get") void getConsumerAssignment(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("consumerGroupId") String consumerGroupId,
      @PathParam("consumerId") String consumerId,
      @PathParam("topicName") String topicName,
      @PathParam("partitionId") int partitionId
  );  }