@Test public void listReplicas_nonExistingPartitionOrTopicOrCluster_throwsNotFound() { expect(replicaManager.listReplicas(CLUSTER_ID, TOPIC_NAME, PARTITION_ID)) .andReturn(failedFuture(new NotFoundException())); replay(replicaManager); FakeAsyncResponse response = new FakeAsyncResponse(); replicasResource.listReplicas(response, CLUSTER_ID, TOPIC_NAME, PARTITION_ID); assertEquals(NotFoundException.class, response.getException().getClass()); }
@GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.list") public void listReplicas( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("topicName") String topicName, @PathParam("partitionId") Integer partitionId ) { CompletableFuture<ListReplicasResponse> response = replicaManager.get() .listReplicas(clusterId, topicName, partitionId) .thenApply( replicas -> ListReplicasResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "topics", topicName, "partitions", Integer.toString(partitionId), "replicas")) .build()) .setData( replicas.stream() .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); }
ReplicasResource { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.list") public void listReplicas( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("topicName") String topicName, @PathParam("partitionId") Integer partitionId ) { CompletableFuture<ListReplicasResponse> response = replicaManager.get() .listReplicas(clusterId, topicName, partitionId) .thenApply( replicas -> ListReplicasResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "topics", topicName, "partitions", Integer.toString(partitionId), "replicas")) .build()) .setData( replicas.stream() .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } }
ReplicasResource { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.list") public void listReplicas( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("topicName") String topicName, @PathParam("partitionId") Integer partitionId ) { CompletableFuture<ListReplicasResponse> response = replicaManager.get() .listReplicas(clusterId, topicName, partitionId) .thenApply( replicas -> ListReplicasResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "topics", topicName, "partitions", Integer.toString(partitionId), "replicas")) .build()) .setData( replicas.stream() .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  ReplicasResource(
      Provider<ReplicaManager> replicaManager, CrnFactory crnFactory, UrlFactory urlFactory); }
ReplicasResource { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.list") public void listReplicas( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("topicName") String topicName, @PathParam("partitionId") Integer partitionId ) { CompletableFuture<ListReplicasResponse> response = replicaManager.get() .listReplicas(clusterId, topicName, partitionId) .thenApply( replicas -> ListReplicasResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "topics", topicName, "partitions", Integer.toString(partitionId), "replicas")) .build()) .setData( replicas.stream() .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  ReplicasResource(
      Provider<ReplicaManager> replicaManager, CrnFactory crnFactory, UrlFactory urlFactory); @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.list") void listReplicas(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("topicName") String topicName,
      @PathParam("partitionId") Integer partitionId
  ); @GET @Path("/{brokerId}") @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.get") void getReplica(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("topicName") String topicName,
      @PathParam("partitionId") Integer partitionId,
      @PathParam("brokerId") Integer brokerId
  ); }
ReplicasResource { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.list") public void listReplicas( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("topicName") String topicName, @PathParam("partitionId") Integer partitionId ) { CompletableFuture<ListReplicasResponse> response = replicaManager.get() .listReplicas(clusterId, topicName, partitionId) .thenApply( replicas -> ListReplicasResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "topics", topicName, "partitions", Integer.toString(partitionId), "replicas")) .build()) .setData( replicas.stream() .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  ReplicasResource(
      Provider<ReplicaManager> replicaManager, CrnFactory crnFactory, UrlFactory urlFactory); @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.list") void listReplicas(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("topicName") String topicName,
      @PathParam("partitionId") Integer partitionId
  ); @GET @Path("/{brokerId}") @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.replicas.get") void getReplica(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("topicName") String topicName,
      @PathParam("partitionId") Integer partitionId,
      @PathParam("brokerId") Integer brokerId
  );  }