class 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) {
        // Retrieve cluster
        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);

        // Create new Operational Client
        try (final KafkaOperations operations = createOperationsClient(cluster)) {
            final boolean result = operations.createTopic(createTopic);

            // Quick n dirty json response
            return new ResultResponse("CreateTopic", result, "Created topic '" + createTopicRequest.getName() + "'");
        } catch (final Exception e) {
            throw new ApiException("CreateTopic", e);
        }
    }

}