@Test public void testGetTopic() { expect(topicManager.getLocalTopic(TOPIC_1.getName())) .andReturn(completedFuture(Optional.of(TOPIC_1))); expect(topicConfigManager.listTopicConfigs(CLUSTER_ID, TOPIC_1.getName())) .andReturn(completedFuture(Arrays.asList(CONFIG_1, CONFIG_2, CONFIG_3))); replay(topicManager, topicConfigManager); Response response1 = request("/topics/" + TOPIC_1.getName(), Versions.KAFKA_V2_JSON).get(); assertOKResponse(response1, Versions.KAFKA_V2_JSON); final GetTopicResponse topicResponse1 = TestUtils.tryReadEntityOrLog(response1, GetTopicResponse.class); assertEquals( GetTopicResponse.fromTopic(TOPIC_1, Arrays.asList(CONFIG_1, CONFIG_2, CONFIG_3)), topicResponse1); }
@GET @Path("/{topic}") @PerformanceMetric("topic.get+v2") public void getTopic( @Suspended AsyncResponse asyncResponse, @PathParam("topic") String topicName) { TopicManager topicManager = topicManagerProvider.get(); TopicConfigManager topicConfigManager = topicConfigManagerProvider.get(); CompletableFuture<Topic> topicFuture = topicManager.getLocalTopic(topicName) .thenApply(topic -> topic.orElseThrow(Errors::topicNotFoundException)); CompletableFuture<GetTopicResponse> response = topicFuture.thenCompose( topic -> topicConfigManager.listTopicConfigs(topic.getClusterId(), topicName)) .thenCombine( topicFuture, (configs, topic) -> GetTopicResponse.fromTopic(topic, configs)); AsyncResponses.asyncResume(asyncResponse, response); }
TopicsResource { @GET @Path("/{topic}") @PerformanceMetric("topic.get+v2") public void getTopic( @Suspended AsyncResponse asyncResponse, @PathParam("topic") String topicName) { TopicManager topicManager = topicManagerProvider.get(); TopicConfigManager topicConfigManager = topicConfigManagerProvider.get(); CompletableFuture<Topic> topicFuture = topicManager.getLocalTopic(topicName) .thenApply(topic -> topic.orElseThrow(Errors::topicNotFoundException)); CompletableFuture<GetTopicResponse> response = topicFuture.thenCompose( topic -> topicConfigManager.listTopicConfigs(topic.getClusterId(), topicName)) .thenCombine( topicFuture, (configs, topic) -> GetTopicResponse.fromTopic(topic, configs)); AsyncResponses.asyncResume(asyncResponse, response); } }
TopicsResource { @GET @Path("/{topic}") @PerformanceMetric("topic.get+v2") public void getTopic( @Suspended AsyncResponse asyncResponse, @PathParam("topic") String topicName) { TopicManager topicManager = topicManagerProvider.get(); TopicConfigManager topicConfigManager = topicConfigManagerProvider.get(); CompletableFuture<Topic> topicFuture = topicManager.getLocalTopic(topicName) .thenApply(topic -> topic.orElseThrow(Errors::topicNotFoundException)); CompletableFuture<GetTopicResponse> response = topicFuture.thenCompose( topic -> topicConfigManager.listTopicConfigs(topic.getClusterId(), topicName)) .thenCombine( topicFuture, (configs, topic) -> GetTopicResponse.fromTopic(topic, configs)); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  TopicsResource(
      Provider<TopicManager> topicManagerProvider,
      Provider<TopicConfigManager> topicConfigManagerProvider); }
TopicsResource { @GET @Path("/{topic}") @PerformanceMetric("topic.get+v2") public void getTopic( @Suspended AsyncResponse asyncResponse, @PathParam("topic") String topicName) { TopicManager topicManager = topicManagerProvider.get(); TopicConfigManager topicConfigManager = topicConfigManagerProvider.get(); CompletableFuture<Topic> topicFuture = topicManager.getLocalTopic(topicName) .thenApply(topic -> topic.orElseThrow(Errors::topicNotFoundException)); CompletableFuture<GetTopicResponse> response = topicFuture.thenCompose( topic -> topicConfigManager.listTopicConfigs(topic.getClusterId(), topicName)) .thenCombine( topicFuture, (configs, topic) -> GetTopicResponse.fromTopic(topic, configs)); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  TopicsResource(
      Provider<TopicManager> topicManagerProvider,
      Provider<TopicConfigManager> topicConfigManagerProvider); @GET @PerformanceMetric("topics.list+v2") void list(@Suspended AsyncResponse asyncResponse); @GET @Path("/{topic}") @PerformanceMetric("topic.get+v2") void getTopic(
      @Suspended AsyncResponse asyncResponse, @PathParam("topic") String topicName); }
TopicsResource { @GET @Path("/{topic}") @PerformanceMetric("topic.get+v2") public void getTopic( @Suspended AsyncResponse asyncResponse, @PathParam("topic") String topicName) { TopicManager topicManager = topicManagerProvider.get(); TopicConfigManager topicConfigManager = topicConfigManagerProvider.get(); CompletableFuture<Topic> topicFuture = topicManager.getLocalTopic(topicName) .thenApply(topic -> topic.orElseThrow(Errors::topicNotFoundException)); CompletableFuture<GetTopicResponse> response = topicFuture.thenCompose( topic -> topicConfigManager.listTopicConfigs(topic.getClusterId(), topicName)) .thenCombine( topicFuture, (configs, topic) -> GetTopicResponse.fromTopic(topic, configs)); AsyncResponses.asyncResume(asyncResponse, response); } @Inject  TopicsResource(
      Provider<TopicManager> topicManagerProvider,
      Provider<TopicConfigManager> topicConfigManagerProvider); @GET @PerformanceMetric("topics.list+v2") void list(@Suspended AsyncResponse asyncResponse); @GET @Path("/{topic}") @PerformanceMetric("topic.get+v2") void getTopic(
      @Suspended AsyncResponse asyncResponse, @PathParam("topic") String topicName);  }