From c4cc22c434a789828bb8e760a1977bb8875c6963 Mon Sep 17 00:00:00 2001
From: Asif Mahmud <asifmahmud.tayef@gmail.com>
Date: Tue, 30 Aug 2022 18:47:21 +0600
Subject: [PATCH] Add option to set topic message retention time in
 topic-detail page

---
 .../kafdrop/controller/TopicController.java   | 18 ++++++++++++
 .../kafdrop/model/UpdateTopicRetentionVO.java | 14 +++++++++
 .../service/KafkaHighLevelAdminClient.java    | 21 ++++++++++++++
 .../java/kafdrop/service/KafkaMonitor.java    |  4 ++-
 .../kafdrop/service/KafkaMonitorImpl.java     |  5 ++++
 .../resources/templates/topic-detail.ftlh     | 29 +++++++++++++++++++
 6 files changed, 90 insertions(+), 1 deletion(-)
 create mode 100644 src/main/java/kafdrop/model/UpdateTopicRetentionVO.java

diff --git a/src/main/java/kafdrop/controller/TopicController.java b/src/main/java/kafdrop/controller/TopicController.java
index 09a946c0..81931ef5 100644
--- a/src/main/java/kafdrop/controller/TopicController.java
+++ b/src/main/java/kafdrop/controller/TopicController.java
@@ -28,6 +28,7 @@
 import org.springframework.stereotype.*;
 import org.springframework.ui.*;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.servlet.ModelAndView;
 
 import java.util.*;
 
@@ -152,4 +153,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 d2c9b859..b1a229f2 100644
--- a/src/main/java/kafdrop/service/KafkaHighLevelAdminClient.java
+++ b/src/main/java/kafdrop/service/KafkaHighLevelAdminClient.java
@@ -153,6 +153,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<AlterConfigOp> alterConfigOps = new ArrayList<>();
+    alterConfigOps.add(config);
+
+    Map<ConfigResource, Collection<AlterConfigOp>> 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<AclBinding> listAcls() {
     final Collection<AclBinding> aclsBindings;
     try {
diff --git a/src/main/java/kafdrop/service/KafkaMonitor.java b/src/main/java/kafdrop/service/KafkaMonitor.java
index 137ac4f3..9a51cd70 100644
--- a/src/main/java/kafdrop/service/KafkaMonitor.java
+++ b/src/main/java/kafdrop/service/KafkaMonitor.java
@@ -60,5 +60,7 @@ List<MessageVO> getMessages(TopicPartition topicPartition, long offset, int coun
    */
   void deleteTopic(String topic);
 
-  List<AclVO> getAcls();
+    void addTopicMessageRetentionPeriodInMs(String topic, int retentionTimeInMs);
+
+    List<AclVO> getAcls();
 }
diff --git a/src/main/java/kafdrop/service/KafkaMonitorImpl.java b/src/main/java/kafdrop/service/KafkaMonitorImpl.java
index f33fd917..0c33bff7 100644
--- a/src/main/java/kafdrop/service/KafkaMonitorImpl.java
+++ b/src/main/java/kafdrop/service/KafkaMonitorImpl.java
@@ -239,6 +239,11 @@ public void deleteTopic(String topic) {
     highLevelAdminClient.deleteTopic(topic);
   }
 
+  @Override
+  public void addTopicMessageRetentionPeriodInMs(String topic, int retentionTimeInMs) {
+    highLevelAdminClient.addTopicMessageRetentionPeriodInMs(topic, retentionTimeInMs);
+  }
+
   @Override
   public List<AclVO> 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 d6ab524f..5119c1cd 100644
--- a/src/main/resources/templates/topic-detail.ftlh
+++ b/src/main/resources/templates/topic-detail.ftlh
@@ -99,6 +99,24 @@
                     </tbody>
                 </table>
             </#if>
+            <form id="update-retention-period-form" action="<@spring.url '/topic/${topic.name}/set-retention-period'/>" method="POST">
+                <table table table-bordered table-sm small style="margin-top: 20px">
+                    <tbody>
+                        <tr>
+                            <td>Retention period in ms </td>
+                        </tr>
+                        <tr>
+                            <td align="center"><input type="number" name="retentionTime" required></td>
+                        </tr>
+                    </tbody>
+                </table>
+                <button style="margin-top: 5px; height: 50%" class="btn btn-success" type="submit"> <i class="fa"></i> Set </button>
+            </form>
+            <#if errorMessage??>
+                <p>Error setting retention period: ${errorMessage}</p>
+            <#elseif topicName??>
+                <p>Successfully set retention period</p>
+            </#if>
         </div>
 
     </div>
@@ -166,6 +184,7 @@
 <script>
   $(document).ready(function () {
     let removalConfirmed = false;
+    let updateRetentionConfirmed = false;
 
     $('#delete-topic-form').submit(function (event) {
       if (!removalConfirmed) {
@@ -176,5 +195,15 @@
         }
       }
     });
+
+    $('#update-retention-period-form').submit(function (event) {
+          if (!updateRetentionConfirmed) {
+            event.preventDefault();
+            if(confirm('Are you sure you want to update the topic retention time?')) {
+              updateRetentionConfirmed = true;
+              $('#update-retention-period-form').submit();
+            }
+          }
+        });
   });
 </script>