@Test public void searchReplicasByBroker_nonExistingBroker_returnsNotFound() { expect(replicaManager.searchReplicasByBrokerId(CLUSTER_ID, BROKER_ID)) .andReturn(failedFuture(new NotFoundException())); replay(replicaManager); FakeAsyncResponse response = new FakeAsyncResponse(); searchReplicasByBrokerAction.searchReplicasByBroker(response, CLUSTER_ID, BROKER_ID); assertEquals(NotFoundException.class, response.getException().getClass()); }
@GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.brokers.partition-replicas.list") public void searchReplicasByBroker( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("brokerId") Integer brokerId ) { CompletableFuture<SearchReplicasByBrokerResponse> response = replicaManager.get() .searchReplicasByBrokerId(clusterId, brokerId) .thenApply( replicas -> SearchReplicasByBrokerResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "brokers", Integer.toString(brokerId), "partition-replicas")) .build()) .setData( replicas.stream() .sorted( comparing(PartitionReplica::getTopicName) .thenComparing(PartitionReplica::getPartitionId)) .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); }
SearchReplicasByBrokerAction { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.brokers.partition-replicas.list") public void searchReplicasByBroker( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("brokerId") Integer brokerId ) { CompletableFuture<SearchReplicasByBrokerResponse> response = replicaManager.get() .searchReplicasByBrokerId(clusterId, brokerId) .thenApply( replicas -> SearchReplicasByBrokerResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "brokers", Integer.toString(brokerId), "partition-replicas")) .build()) .setData( replicas.stream() .sorted( comparing(PartitionReplica::getTopicName) .thenComparing(PartitionReplica::getPartitionId)) .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } }
SearchReplicasByBrokerAction { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.brokers.partition-replicas.list") public void searchReplicasByBroker( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("brokerId") Integer brokerId ) { CompletableFuture<SearchReplicasByBrokerResponse> response = replicaManager.get() .searchReplicasByBrokerId(clusterId, brokerId) .thenApply( replicas -> SearchReplicasByBrokerResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "brokers", Integer.toString(brokerId), "partition-replicas")) .build()) .setData( replicas.stream() .sorted( comparing(PartitionReplica::getTopicName) .thenComparing(PartitionReplica::getPartitionId)) .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  SearchReplicasByBrokerAction(
      Provider<ReplicaManager> replicaManager, CrnFactory crnFactory, UrlFactory urlFactory); }
SearchReplicasByBrokerAction { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.brokers.partition-replicas.list") public void searchReplicasByBroker( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("brokerId") Integer brokerId ) { CompletableFuture<SearchReplicasByBrokerResponse> response = replicaManager.get() .searchReplicasByBrokerId(clusterId, brokerId) .thenApply( replicas -> SearchReplicasByBrokerResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "brokers", Integer.toString(brokerId), "partition-replicas")) .build()) .setData( replicas.stream() .sorted( comparing(PartitionReplica::getTopicName) .thenComparing(PartitionReplica::getPartitionId)) .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  SearchReplicasByBrokerAction(
      Provider<ReplicaManager> replicaManager, CrnFactory crnFactory, UrlFactory urlFactory); @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.brokers.partition-replicas.list") void searchReplicasByBroker(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("brokerId") Integer brokerId
  ); }
SearchReplicasByBrokerAction { @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.brokers.partition-replicas.list") public void searchReplicasByBroker( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @PathParam("brokerId") Integer brokerId ) { CompletableFuture<SearchReplicasByBrokerResponse> response = replicaManager.get() .searchReplicasByBrokerId(clusterId, brokerId) .thenApply( replicas -> SearchReplicasByBrokerResponse.create( ReplicaDataList.builder() .setMetadata( ResourceCollection.Metadata.builder() .setSelf( urlFactory.create( "v3", "clusters", clusterId, "brokers", Integer.toString(brokerId), "partition-replicas")) .build()) .setData( replicas.stream() .sorted( comparing(PartitionReplica::getTopicName) .thenComparing(PartitionReplica::getPartitionId)) .map(this::toReplicaData) .collect(Collectors.toList())) .build())); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  SearchReplicasByBrokerAction(
      Provider<ReplicaManager> replicaManager, CrnFactory crnFactory, UrlFactory urlFactory); @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.brokers.partition-replicas.list") void searchReplicasByBroker(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("brokerId") Integer brokerId
  );  }