ApiController extends BaseController { @ResponseBody @RequestMapping(path = "/cluster/{id}/create/topic", method = RequestMethod.POST, produces = "application/json") public ResultResponse createTopic(@PathVariable final Long id, @RequestBody final CreateTopicRequest createTopicRequest) { final Cluster cluster = retrieveClusterById(id); final String name = createTopicRequest.getName(); if (name == null || name.trim().isEmpty()) { throw new ApiException("CreateTopic", "Invalid topic name"); } final Integer partitions = createTopicRequest.getPartitions(); if (partitions == null || partitions < 1) { throw new ApiException("CreateTopic", "Invalid partitions value"); } final Short replicas = createTopicRequest.getReplicas(); if (replicas == null || replicas < 1) { throw new ApiException("CreateTopic", "Invalid replicas value"); } final CreateTopic createTopic = new CreateTopic(name, partitions, replicas); try (final KafkaOperations operations = createOperationsClient(cluster)) { final boolean result = operations.createTopic(createTopic); return new ResultResponse("CreateTopic", result, "Created topic '" + createTopicRequest.getName() + "'"); } catch (final Exception e) { throw new ApiException("CreateTopic", e); } }  @ResponseBody @RequestMapping(path = "/consumer/view/{id}", method = RequestMethod.POST, produces = "application/json") KafkaResults consume(
        @PathVariable final Long id,
        @RequestBody final ConsumeRequest consumeRequest); @ResponseBody @RequestMapping( path = "/consumer/view/{id}/offsets", method = RequestMethod.POST, produces = "application/json" ) ConsumerState setConsumerOffsets(
        @PathVariable final Long id,
        @RequestBody final Map<Integer, Long> partitionOffsetMap
    ); @ResponseBody @RequestMapping( path = "/consumer/view/{id}/timestamp/{timestamp}", method = RequestMethod.POST, produces = "application/json" ) ConsumerState setConsumerOffsetsByTimestamp(
        @PathVariable final Long id,
        @PathVariable final Long timestamp
    ); @ResponseBody @RequestMapping(path = "/view/{id}/partitions", method = RequestMethod.GET, produces = "application/json") Collection<Integer> getPartitionsForView(@PathVariable final Long id); @ResponseBody @RequestMapping(path = "/cluster/{id}/topics/list", method = RequestMethod.GET, produces = "application/json") List<TopicListing> getTopics(@PathVariable final Long id, @RequestParam(required = false) final String search); @ResponseBody @RequestMapping(path = "/cluster/{id}/topic/{topic}/details", method = RequestMethod.GET, produces = "application/json") TopicDetails getTopicDetails(@PathVariable final Long id, @PathVariable final String topic); @ResponseBody @RequestMapping(path = "/cluster/{id}/topic/{topic}/config", method = RequestMethod.GET, produces = "application/json") List<ConfigItem> getTopicConfig(@PathVariable final Long id, @PathVariable final String topic); @ResponseBody @RequestMapping(path = "/cluster/{id}/broker/{brokerId}/config", method = RequestMethod.GET, produces = "application/json") List<ConfigItem> getBrokerConfig(@PathVariable final Long id, @PathVariable final String brokerId); @ResponseBody @RequestMapping(path = "/cluster/{id}/topics/details", method = RequestMethod.GET, produces = "application/json") Collection<TopicDetails> getAllTopicsDetails(@PathVariable final Long id); @ResponseBody @RequestMapping(path = "/cluster/{id}/create/topic", method = RequestMethod.POST, produces = "application/json") ResultResponse createTopic(@PathVariable final Long id, @RequestBody final CreateTopicRequest createTopicRequest); @ResponseBody @RequestMapping(path = "/cluster/{id}/modify/topic", method = RequestMethod.POST, produces = "application/json") List<ConfigItem> modifyTopicConfig(
        @PathVariable final Long id,
        @RequestBody final ModifyTopicConfigRequest modifyTopicConfigRequest
    ); ResultResponse deleteTopic(@PathVariable final Long id, @RequestBody final DeleteTopicRequest deleteTopicRequest); @ResponseBody @RequestMapping(path = "/cluster/{id}/nodes", method = RequestMethod.GET, produces = "application/json") List<NodeDetails> getClusterNodes(@PathVariable final Long id); @ResponseBody @RequestMapping(path = "/filter/{id}/options", method = RequestMethod.GET, produces = "application/json") String[] getFilterOptions(@PathVariable final Long id); @ResponseBody @RequestMapping(path = "/cluster/{id}/consumers", method = RequestMethod.GET, produces = "application/json") List<ConsumerGroupIdentifier> listConsumers(@PathVariable final Long id); @ResponseBody @RequestMapping( path = "/cluster/{id}/consumersAndDetails", method = RequestMethod.GET, produces = "application/json" ) List<ConsumerGroupDetails> listConsumersAndDetails(@PathVariable final Long id); @ResponseBody @RequestMapping( path = "/cluster/{id}/consumer/{consumerGroupId}/details", method = RequestMethod.GET, produces = "application/json" ) ConsumerGroupDetails getConsumerDetails(
        @PathVariable final Long id,
        @PathVariable final String consumerGroupId
    ); @ResponseBody @RequestMapping( path = "/cluster/{id}/consumer/{consumerGroupId}/offsets", method = RequestMethod.GET, produces = "application/json" ) ConsumerGroupOffsets getConsumerOffsets(
        @PathVariable final Long id,
        @PathVariable final String consumerGroupId
    ); @ResponseBody @RequestMapping( path = "/cluster/{id}/consumer/{consumerGroupId}/offsetsAndTailPositions", method = RequestMethod.GET, produces = "application/json" ) ConsumerGroupOffsetsWithTailPositions getConsumerOffsetsWithTailPositions(
        @PathVariable final Long id,
        @PathVariable final String consumerGroupId
    ); @ResponseBody @RequestMapping(path = "/cluster/{id}/consumer/remove", method = RequestMethod.POST, produces = "application/json") boolean removeConsumer(
        @PathVariable final Long id,
        @RequestBody final ConsumerRemoveRequest consumerRemoveRequest); @ResponseBody @ExceptionHandler(ApiException.class) @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR) ApiErrorResponse handleApiException(final ApiException exception); @Override @ModelAttribute void addAttributes(final Model model);  }
@Test @Transactional public void test_modifyTopic() throws Exception { final String topicName = "TestTopic-" + System.currentTimeMillis(); final String configName1 = "flush.messages"; final String newConfigValue1 = "0"; final String configName2 = "max.message.bytes"; final String newConfigValue2 = "1024"; final Cluster cluster = clusterTestTools.createCluster( "Test Cluster", sharedKafkaTestResource.getKafkaConnectString() ); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 1, (short) 1); final String payload = "{ \"topic\": \"" + topicName + "\", \"config\": {" + "\"" + configName1 + "\": \"" + newConfigValue1 + "\", " + "\"" + configName2 + "\": \"" + newConfigValue2 + "\"" + "}}"; final MvcResult result = mockMvc .perform(post("/api/cluster/" + cluster.getId() + "/modify/topic") .with(user(adminUserDetails)) .with(csrf()) .content(payload) .contentType(MediaType.APPLICATION_JSON) ) .andDo(print()) .andExpect(status().isOk()) .andReturn(); final String resultJsonStr = result.getResponse().getContentAsString(); final String targetItem1 = "{\"name\":\"" + configName1 + "\",\"value\":\"" + newConfigValue1 + "\",\"default\":false}"; final String targetItem2 = "{\"name\":\"" + configName2 + "\",\"value\":\"" + newConfigValue2 + "\",\"default\":false}"; assertTrue(resultJsonStr.contains(targetItem1)); assertTrue(resultJsonStr.contains(targetItem2)); }
@Test @Transactional public void test_listTopics_noSearchString() throws Exception { final Cluster cluster = clusterTestTools.createCluster( "Test Cluster", sharedKafkaTestResource.getKafkaConnectString() ); final String expectedTopic1 = "A-ExpectedTopic1-" + System.currentTimeMillis(); final String expectedTopic2= "C-ExpectedTopic2-" + System.currentTimeMillis(); final String expectedTopic3 = "B-ExpectedTopic3-" + System.currentTimeMillis(); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(expectedTopic1, 1, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(expectedTopic2, 1, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(expectedTopic3, 1, (short) 1); mockMvc .perform(get("/api/cluster/" + cluster.getId() + "/topics/list") .with(user(nonAdminUserDetails)) .with(csrf()) .contentType(MediaType.APPLICATION_JSON) ) .andDo(print()) .andExpect(status().isOk()) .andExpect(content().string(containsString(expectedTopic1))) .andExpect(content().string(containsString(expectedTopic2))) .andExpect(content().string(containsString(expectedTopic3))); }
@Test @Transactional public void test_listTopics_withSearchString() throws Exception { final Cluster cluster = clusterTestTools.createCluster( "Test Cluster", sharedKafkaTestResource.getKafkaConnectString() ); final String searchStr = "CaT"; final String expectedTopic1 = "ExpectedTopic1-" + System.currentTimeMillis(); final String expectedTopic2= "ExpectedCATTopic2-" + System.currentTimeMillis(); final String expectedTopic3 = "ExpectedTopic3-" + System.currentTimeMillis(); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(expectedTopic1, 1, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(expectedTopic2, 1, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(expectedTopic3, 1, (short) 1); mockMvc .perform(get("/api/cluster/" + cluster.getId() + "/topics/list") .param("search", searchStr) .with(user(nonAdminUserDetails)) .with(csrf()) .contentType(MediaType.APPLICATION_JSON) ) .andDo(print()) .andExpect(status().isOk()) .andExpect(content().string(containsString(expectedTopic2))) .andExpect(content().string(Matchers.not(containsString(expectedTopic1)))) .andExpect(content().string(Matchers.not(containsString(expectedTopic3)))); }