@Test public void createTopic_existingTopic_throwsTopicExists() { expect( topicManager.createTopic( CLUSTER_ID, TOPIC_1.getName(), TOPIC_1.getPartitions().size(), TOPIC_1.getReplicationFactor(), singletonMap("cleanup.policy", "compact"))) .andReturn(failedFuture(new TopicExistsException(""))); replay(topicManager); FakeAsyncResponse response = new FakeAsyncResponse(); topicsResource.createTopic( response, TOPIC_1.getClusterId(), CreateTopicRequest.builder() .setTopicName(TOPIC_1.getName()) .setPartitionsCount(TOPIC_1.getPartitions().size()) .setReplicationFactor(TOPIC_1.getReplicationFactor()) .setConfigs( singletonList(CreateTopicRequest.ConfigEntry.create("cleanup.policy", "compact"))) .build()); assertEquals(TopicExistsException.class, response.getException().getClass()); }
@POST @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.create") public void createTopic( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @Valid CreateTopicRequest request ) { String topicName = request.getTopicName(); int partitionsCount = request.getPartitionsCount(); short replicationFactor = request.getReplicationFactor(); Map<String, String> configs = new HashMap<>(); request.getConfigs() .forEach(entry -> configs.put(entry.getName(), entry.getValue().orElse(null))); TopicData topicData = toTopicData( Topic.create( clusterId, topicName, emptyList(), replicationFactor, false)); CompletableFuture<CreateTopicResponse> response = topicManager.get() .createTopic(clusterId, topicName, partitionsCount, replicationFactor, configs) .thenApply(none -> CreateTopicResponse.create(topicData)); AsyncResponseBuilder.from( Response.status(Status.CREATED).location(URI.create(topicData.getMetadata().getSelf()))) .entity(response) .asyncResume(asyncResponse); }
TopicsResource { @POST @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.create") public void createTopic( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @Valid CreateTopicRequest request ) { String topicName = request.getTopicName(); int partitionsCount = request.getPartitionsCount(); short replicationFactor = request.getReplicationFactor(); Map<String, String> configs = new HashMap<>(); request.getConfigs() .forEach(entry -> configs.put(entry.getName(), entry.getValue().orElse(null))); TopicData topicData = toTopicData( Topic.create( clusterId, topicName, emptyList(), replicationFactor, false)); CompletableFuture<CreateTopicResponse> response = topicManager.get() .createTopic(clusterId, topicName, partitionsCount, replicationFactor, configs) .thenApply(none -> CreateTopicResponse.create(topicData)); AsyncResponseBuilder.from( Response.status(Status.CREATED).location(URI.create(topicData.getMetadata().getSelf()))) .entity(response) .asyncResume(asyncResponse); } }
TopicsResource { @POST @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.create") public void createTopic( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @Valid CreateTopicRequest request ) { String topicName = request.getTopicName(); int partitionsCount = request.getPartitionsCount(); short replicationFactor = request.getReplicationFactor(); Map<String, String> configs = new HashMap<>(); request.getConfigs() .forEach(entry -> configs.put(entry.getName(), entry.getValue().orElse(null))); TopicData topicData = toTopicData( Topic.create( clusterId, topicName, emptyList(), replicationFactor, false)); CompletableFuture<CreateTopicResponse> response = topicManager.get() .createTopic(clusterId, topicName, partitionsCount, replicationFactor, configs) .thenApply(none -> CreateTopicResponse.create(topicData)); AsyncResponseBuilder.from( Response.status(Status.CREATED).location(URI.create(topicData.getMetadata().getSelf()))) .entity(response) .asyncResume(asyncResponse); } @Inject  TopicsResource(
      Provider<TopicManager> topicManager,
      CrnFactory crnFactory,
      UrlFactory urlFactory
  ); }
TopicsResource { @POST @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.create") public void createTopic( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @Valid CreateTopicRequest request ) { String topicName = request.getTopicName(); int partitionsCount = request.getPartitionsCount(); short replicationFactor = request.getReplicationFactor(); Map<String, String> configs = new HashMap<>(); request.getConfigs() .forEach(entry -> configs.put(entry.getName(), entry.getValue().orElse(null))); TopicData topicData = toTopicData( Topic.create( clusterId, topicName, emptyList(), replicationFactor, false)); CompletableFuture<CreateTopicResponse> response = topicManager.get() .createTopic(clusterId, topicName, partitionsCount, replicationFactor, configs) .thenApply(none -> CreateTopicResponse.create(topicData)); AsyncResponseBuilder.from( Response.status(Status.CREATED).location(URI.create(topicData.getMetadata().getSelf()))) .entity(response) .asyncResume(asyncResponse); } @Inject  TopicsResource(
      Provider<TopicManager> topicManager,
      CrnFactory crnFactory,
      UrlFactory urlFactory
  ); @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.list") void listTopics(
      @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId); @GET @Path("/{topicName}") @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.get") void getTopic(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("topicName") String topicName
  ); @POST @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.create") void createTopic(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @Valid CreateTopicRequest request
  ); @DELETE @Path("/{topicName}") @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.delete") void deleteTopic(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("topicName") String topicName
  ); }
TopicsResource { @POST @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.create") public void createTopic( @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId, @Valid CreateTopicRequest request ) { String topicName = request.getTopicName(); int partitionsCount = request.getPartitionsCount(); short replicationFactor = request.getReplicationFactor(); Map<String, String> configs = new HashMap<>(); request.getConfigs() .forEach(entry -> configs.put(entry.getName(), entry.getValue().orElse(null))); TopicData topicData = toTopicData( Topic.create( clusterId, topicName, emptyList(), replicationFactor, false)); CompletableFuture<CreateTopicResponse> response = topicManager.get() .createTopic(clusterId, topicName, partitionsCount, replicationFactor, configs) .thenApply(none -> CreateTopicResponse.create(topicData)); AsyncResponseBuilder.from( Response.status(Status.CREATED).location(URI.create(topicData.getMetadata().getSelf()))) .entity(response) .asyncResume(asyncResponse); } @Inject  TopicsResource(
      Provider<TopicManager> topicManager,
      CrnFactory crnFactory,
      UrlFactory urlFactory
  ); @GET @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.list") void listTopics(
      @Suspended AsyncResponse asyncResponse, @PathParam("clusterId") String clusterId); @GET @Path("/{topicName}") @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.get") void getTopic(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("topicName") String topicName
  ); @POST @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.create") void createTopic(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @Valid CreateTopicRequest request
  ); @DELETE @Path("/{topicName}") @Produces(MediaType.APPLICATION_JSON) @PerformanceMetric("v3.topics.delete") void deleteTopic(
      @Suspended AsyncResponse asyncResponse,
      @PathParam("clusterId") String clusterId,
      @PathParam("topicName") String topicName
  );  }