Allow searching messages in a specific partition (#678)

obsidiandynamics · Nov 26, 2024 · d6d43a7 · d6d43a7
1 parent 10d4537
commit d6d43a7
Show file tree

Hide file tree

Showing 6 changed files with 38 additions and 5 deletions.
diff --git a/src/main/java/kafdrop/controller/MessageController.java b/src/main/java/kafdrop/controller/MessageController.java
@@ -218,6 +218,7 @@ public String searchMessageForm(@PathVariable("name") String topicName,
       defaultForm.setSearchText("");
       defaultForm.setFormat(defaultFormat);
       defaultForm.setKeyFormat(defaultKeyFormat);
+      defaultForm.setPartition(-1);
       defaultForm.setMaximumCount(100);
       defaultForm.setStartTimestamp(new Date(0));
       model.addAttribute("searchMessageForm", defaultForm);
@@ -231,6 +232,7 @@ public String searchMessageForm(@PathVariable("name") String topicName,
     model.addAttribute("messageFormats", MessageFormat.values());
     model.addAttribute("defaultKeyFormat", defaultKeyFormat);
     model.addAttribute("keyFormats", KeyFormat.values());
+    model.addAttribute("partitions", topic.getPartitions().stream().map(TopicPartitionVO::getId).toList());
     model.addAttribute("descFiles", protobufProperties.getDescFilesList());
 
     if (!searchMessageForm.isEmpty() && !errors.hasErrors()) {
@@ -247,6 +249,7 @@ public String searchMessageForm(@PathVariable("name") String topicName,
       var searchResults = kafkaMonitor.searchMessages(
         topicName,
         searchMessageForm.getSearchText(),
+        searchMessageForm.getPartition(),
         searchMessageForm.getMaximumCount(),
         searchMessageForm.getStartTimestamp(),
         deserializers);

diff --git a/src/main/java/kafdrop/form/SearchMessageForm.java b/src/main/java/kafdrop/form/SearchMessageForm.java
@@ -20,6 +20,8 @@ public class SearchMessageForm {
   @Max(1000)
   private Integer maximumCount;
 
+  private Integer partition;
+
   private MessageFormat format;
 
   private MessageFormat keyFormat;
@@ -103,4 +105,12 @@ public String getMsgTypeName() {
   public void setMsgTypeName(String msgTypeName) {
     this.msgTypeName = msgTypeName;
   }
+
+  public Integer getPartition() {
+    return partition;
+  }
+
+  public void setPartition(Integer partition) {
+    this.partition = partition;
+  }
 }
diff --git a/src/main/java/kafdrop/service/KafkaHighLevelConsumer.java b/src/main/java/kafdrop/service/KafkaHighLevelConsumer.java
@@ -204,11 +204,19 @@ synchronized List<ConsumerRecord<String, String>> getLatestRecords(String topic,
    */
   synchronized SearchResults searchRecords(String topic,
                                            String searchString,
+                                           Integer partition,
                                            Integer maximumCount,
                                            Date startTimestamp,
                                            Deserializers deserializers) {
     initializeClient();
-    final List<TopicPartition> partitions = determinePartitionsForTopic(topic);
+    List<TopicPartition> partitions = determinePartitionsForTopic(topic);
+    if (partition != -1) {
+      var partitionOpt = partitions.stream().filter(p -> p.partition() == partition).findAny();
+      if (partitionOpt.isEmpty()) {
+        throw new IllegalArgumentException("Partition does not exist in topic");
+      }
+      partitions = List.of(partitionOpt.get());
+    }
     kafkaConsumer.assign(partitions);
     seekToTimestamp(partitions, startTimestamp);
 

diff --git a/src/main/java/kafdrop/service/KafkaMonitor.java b/src/main/java/kafdrop/service/KafkaMonitor.java
@@ -60,6 +60,7 @@ List<MessageVO> getMessages(TopicPartition topicPartition, long offset, int coun
 
   SearchResultsVO searchMessages(String topic,
                                  String searchString,
+                                 Integer partition,
                                  Integer maximumCount,
                                  Date startTimestamp,
                                  Deserializers deserializers);

diff --git a/src/main/java/kafdrop/service/KafkaMonitorImpl.java b/src/main/java/kafdrop/service/KafkaMonitorImpl.java
@@ -250,10 +250,11 @@ public List<ConsumerVO> getConsumersByTopics(Collection<TopicVO> topicVos) {
   @Override
   public SearchResultsVO searchMessages(String topic,
                                         String searchString,
-                                        Integer maxmuimCount,
+                                        Integer partition,
+                                        Integer maximumCount,
                                         Date startTimestamp,
                                         Deserializers deserializers) {
-    final var records = highLevelConsumer.searchRecords(topic, searchString, maxmuimCount, startTimestamp,
+    final var records = highLevelConsumer.searchRecords(topic, searchString, partition, maximumCount, startTimestamp,
       deserializers);
     final var results = new SearchResultsVO();
 

diff --git a/src/main/resources/templates/search-message.ftlh b/src/main/resources/templates/search-message.ftlh
@@ -70,13 +70,23 @@
     <form method="GET" action="<@spring.url '/topic/${topic.name}/search-messages'/>" id="searchMessageForm" class="card-body">
         <div class="row g-3">
             <@spring.bind path="searchMessageForm.searchText"/>
-            <div class="col-7 form-group ${spring.status.error?string("has-error", "")}">
+            <div class="col-5 form-group ${spring.status.error?string("has-error", "")}">
                 <label class="control-label" for="searchText">Search text</label>
                 <@spring.formInput path="searchMessageForm.searchText" attributes='class="form-control" size="40"'/>
                 <#if spring.status.error>
                     <span class="text-danger"><i class="fa fa-times-circle"></i><@spring.showErrors "<br/>"/></span>
                 </#if>
             </div>
+            <@spring.bind path="searchMessageForm.partition"/>
+            <div class="col-2 form-group ${spring.status.error?string("has-error", "")}">
+                <label class="control-label" for="partition">Partition</label>
+                <select class="form-control" id="partition" name="partition">
+                    <option value="-1">All</option>
+                    <#list partitions as p>
+                        <option value="${p}">${p}</option>
+                    </#list>
+                </select>
+            </div>
             <@spring.bind path="searchMessageForm.startTimestamp"/>
             <div class="col form-group ${spring.status.error?string("has-error", "")}">
                 <label class=control-label" for="maximumCount">Start Timestamp</label>
@@ -169,4 +179,4 @@
     </#if>
 </div>
 
-<@template.footer/>
+<@template.footer/>