diff --git a/src/main/java/kafdrop/controller/TopicController.java b/src/main/java/kafdrop/controller/TopicController.java index 91ed718e..21f892a2 100644 --- a/src/main/java/kafdrop/controller/TopicController.java +++ b/src/main/java/kafdrop/controller/TopicController.java @@ -30,14 +30,14 @@ import kafdrop.service.TopicNotFoundException; import kafdrop.util.MessageFormat; import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.*; +import org.springframework.stereotype.*; +import org.springframework.ui.*; +import org.springframework.web.bind.annotation.*; +import org.springframework.web.servlet.ModelAndView; import org.springframework.http.MediaType; import org.springframework.stereotype.Controller; import org.springframework.ui.Model; -import org.springframework.web.bind.annotation.GetMapping; -import org.springframework.web.bind.annotation.PathVariable; -import org.springframework.web.bind.annotation.PostMapping; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.ResponseBody; import java.util.Collections; import java.util.List; @@ -162,4 +162,21 @@ public String createTopic(CreateTopicVO createTopicVO, Model model) { model.addAttribute("brokersCount", kafkaMonitor.getBrokers().size()); return "topic-create"; } + + @ApiOperation(value = "createTopic", notes = "Create topic") + @ApiResponses(value = { + @ApiResponse(code = 200, message = "Success", response = String.class) + }) + @PostMapping(path = "/{name:.+}/set-retention-period", produces = MediaType.APPLICATION_JSON_VALUE) + public ModelAndView createTopic(@PathVariable("name") String topicName, + UpdateTopicRetentionVO updateTopicRetentionVO, + Model model) { + try { + kafkaMonitor.addTopicMessageRetentionPeriodInMs(topicName, updateTopicRetentionVO.getRetentionTime()); + } catch (Exception ex) { + model.addAttribute("errorMessage", ex.getMessage()); + } + + return new ModelAndView("redirect:/topic/" + topicName); + } } diff --git a/src/main/java/kafdrop/model/UpdateTopicRetentionVO.java b/src/main/java/kafdrop/model/UpdateTopicRetentionVO.java new file mode 100644 index 00000000..90fed95e --- /dev/null +++ b/src/main/java/kafdrop/model/UpdateTopicRetentionVO.java @@ -0,0 +1,14 @@ +package kafdrop.model; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiParam; +import lombok.Data; +import lombok.RequiredArgsConstructor; + +@Data +@RequiredArgsConstructor +@ApiModel("Update topic retention model") +public final class UpdateTopicRetentionVO { + @ApiParam("Retention period in ms") + int retentionTime; +} diff --git a/src/main/java/kafdrop/service/KafkaHighLevelAdminClient.java b/src/main/java/kafdrop/service/KafkaHighLevelAdminClient.java index f9e572bf..cfc2c18f 100644 --- a/src/main/java/kafdrop/service/KafkaHighLevelAdminClient.java +++ b/src/main/java/kafdrop/service/KafkaHighLevelAdminClient.java @@ -167,6 +167,27 @@ void deleteTopic(String topic) { } } + void addTopicMessageRetentionPeriodInMs(String topic, int retentionTimeInMs) { + ConfigResource resource = new ConfigResource(Type.TOPIC, topic); + + ConfigEntry configEntry = new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionTimeInMs)); + AlterConfigOp config = new AlterConfigOp(configEntry, AlterConfigOp.OpType.SET); + + Collection alterConfigOps = new ArrayList<>(); + alterConfigOps.add(config); + + Map> configs = new HashMap<>(); + configs.put(resource, alterConfigOps); + + try { + adminClient.incrementalAlterConfigs(configs); + LOG.info("Topic {} retention time successfully set to {}ms", topic, retentionTimeInMs); + } catch (Exception e) { + LOG.error("Error while update retention period topic", e); + throw new KafkaAdminClientException(e); + } + } + Collection listAcls() { final Collection aclsBindings; try { diff --git a/src/main/java/kafdrop/service/KafkaMonitor.java b/src/main/java/kafdrop/service/KafkaMonitor.java index c45a195a..35760058 100644 --- a/src/main/java/kafdrop/service/KafkaMonitor.java +++ b/src/main/java/kafdrop/service/KafkaMonitor.java @@ -78,5 +78,7 @@ SearchResultsVO searchMessages(String topic, */ void deleteTopic(String topic); - List getAcls(); + void addTopicMessageRetentionPeriodInMs(String topic, int retentionTimeInMs); + + List getAcls(); } diff --git a/src/main/java/kafdrop/service/KafkaMonitorImpl.java b/src/main/java/kafdrop/service/KafkaMonitorImpl.java index 36145799..f231a122 100644 --- a/src/main/java/kafdrop/service/KafkaMonitorImpl.java +++ b/src/main/java/kafdrop/service/KafkaMonitorImpl.java @@ -314,6 +314,11 @@ public void deleteTopic(String topic) { highLevelAdminClient.deleteTopic(topic); } + @Override + public void addTopicMessageRetentionPeriodInMs(String topic, int retentionTimeInMs) { + highLevelAdminClient.addTopicMessageRetentionPeriodInMs(topic, retentionTimeInMs); + } + @Override public List getAcls() { final var acls = highLevelAdminClient.listAcls(); diff --git a/src/main/resources/templates/topic-detail.ftlh b/src/main/resources/templates/topic-detail.ftlh index db9c12b6..49b3d8e1 100644 --- a/src/main/resources/templates/topic-detail.ftlh +++ b/src/main/resources/templates/topic-detail.ftlh @@ -103,6 +103,24 @@ +
+ + + + + + + + + +
Retention period in ms
+ +
+ <#if errorMessage??> +

Error setting retention period: ${errorMessage}

+ <#elseif topicName??> +

Successfully set retention period

+ @@ -170,6 +188,7 @@