[ISSUE-142] Add ability to search brokers, configs, topics (#150)

* Add backend changes to allow searching topics

* Add backend code to support removing topics

* Add initial UI to remove topics

* Better UI for confirming topic removal

* First pass search over topics

* First pass search over topics

* Update version and CHANGELOG

* fix changelog

* Disable topic deletion for now

* Slightly refactor search logic, add search to consumers

* More searching

* More searching

* Update changelog

Author: Crim

CHANGELOG.md

@@ -2,12 +2,13 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
-## 2.1.5 (UNRELEASED)
+## 2.2.0 (UNRELEASED)
 
 #### Bug fixes
 - [ISSUE-143](https://github.com/SourceLabOrg/kafka-webview/issues/143) Fix URLs for stream connections when running Kafka-Webview behind a reverse proxy with a URL Prefix.
 
-#### New features
+#### New Features
+- Ability to search various datatables within the Cluster section of the application.
 
 ## 2.1.4 (02/19/2019)
 

dev-cluster/pom.xml

@@ -5,12 +5,12 @@
     <parent>
         <artifactId>kafka-webview</artifactId>
         <groupId>org.sourcelab</groupId>
-        <version>2.1.5</version>
+        <version>2.2.0</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>dev-cluster</artifactId>
-    <version>2.1.5</version>
+    <version>2.2.0</version>
 
     <!-- Require Maven 3.3.9 -->
     <prerequisites>
@@ -127,6 +127,4 @@
             </plugin>
         </plugins>
     </build>
-
-
 </project>

kafka-webview-plugin/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.sourcelab</groupId>
         <artifactId>kafka-webview</artifactId>
-        <version>2.1.5</version>
+        <version>2.2.0</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>kafka-webview-plugin</artifactId>

kafka-webview-ui/pom.xml

@@ -5,11 +5,11 @@
     <parent>
         <artifactId>kafka-webview</artifactId>
         <groupId>org.sourcelab</groupId>
-        <version>2.1.5</version>
+        <version>2.2.0</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>kafka-webview-ui</artifactId>
-    <version>2.1.5</version>
+    <version>2.2.0</version>
 
     <!-- Module Description and Ownership -->
     <name>Kafka WebView UI</name>

kafka-webview-ui/src/main/frontend/js/app.js

@@ -326,6 +326,23 @@ var ApiClient = {
             }
         });
     },
+    removeTopic: function(clusterId, name, callback) {
+        var payload = JSON.stringify({
+            name: name
+        });
+        jQuery.ajax({
+            type: 'POST',
+            url: '/api/cluster/' + clusterId + '/delete/topic',
+            data: payload,
+            dataType: 'json',
+            headers: ApiClient.getCsrfHeader(),
+            success: callback,
+            error: ApiClient.defaultErrorHandler,
+            beforeSend: function(xhr) {
+                xhr.setRequestHeader('Content-type', 'application/json; charset=utf-8');
+            }
+        });
+    },
     modifyTopicConfig: function(clusterId, topic, configName, configValue, callback) {
         var payloadJson = {
             topic: topic,
@@ -454,3 +471,28 @@ var DateTools = {
         return timezone_standard;
     }
 };
+
+/**
+ * Common Search Tooling.
+ */
+var SearchTools = {
+    doesMatchText : function(searchStr, content) {
+        // If empty search String, assume match all.
+        if (searchStr === null || searchStr.length === 0) {
+            return true;
+        }
+
+        // If content is null, cannot match.
+        if (content === null) {
+            return false;
+        }
+
+        // Otherwise check to see if it matches
+        if (content.toLowerCase().indexOf(searchStr.toLowerCase()) === -1) {
+            // Doesn't match
+            return false;
+        }
+        // Matches
+        return true;
+    }
+};

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/configuration/SecurityConfig.java

@@ -196,6 +196,9 @@ private void enableUserAuth(final HttpSecurity http) throws Exception {
                 // Modify topic
                 "/api/cluster/*/modify/**",
 
+                // Delete topic
+                "/api/cluster/*/delete/**",
+
                 // Remove consumer group
                 "/api/cluster/*/consumer/remove"
 

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/ApiController.java

@@ -30,6 +30,7 @@
 import org.sourcelab.kafka.webview.ui.controller.api.requests.ConsumeRequest;
 import org.sourcelab.kafka.webview.ui.controller.api.requests.ConsumerRemoveRequest;
 import org.sourcelab.kafka.webview.ui.controller.api.requests.CreateTopicRequest;
+import org.sourcelab.kafka.webview.ui.controller.api.requests.DeleteTopicRequest;
 import org.sourcelab.kafka.webview.ui.controller.api.requests.ModifyTopicConfigRequest;
 import org.sourcelab.kafka.webview.ui.controller.api.responses.ResultResponse;
 import org.sourcelab.kafka.webview.ui.manager.kafka.KafkaOperations;
@@ -70,6 +71,7 @@
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.bind.annotation.ResponseStatus;
 
@@ -225,13 +227,22 @@ public Collection<Integer> getPartitionsForView(@PathVariable final Long id) {
      */
     @ResponseBody
     @RequestMapping(path = "/cluster/{id}/topics/list", method = RequestMethod.GET, produces = "application/json")
-    public List<TopicListing> getTopics(@PathVariable final Long id) {
+    public List<TopicListing> getTopics(@PathVariable final Long id, @RequestParam(required = false) final String search) {
         // Retrieve cluster
         final Cluster cluster = retrieveClusterById(id);
 
         // Create new Operational Client
         try (final KafkaOperations operations = createOperationsClient(cluster)) {
-            final TopicList topics = operations.getAvailableTopics();
+            // Get all topics available on cluster.
+            TopicList topics = operations.getAvailableTopics();
+
+            // If search value supplied
+            if (search != null && !search.trim().isEmpty()) {
+                // filter
+                topics = topics.filterByTopicName(search);
+            }
+
+            // return matched topics.
             return topics.getTopics();
         } catch (final Exception e) {
             throw new ApiException("Topics", e);
@@ -349,7 +360,7 @@ public ResultResponse createTopic(@PathVariable final Long id, @RequestBody fina
             final boolean result = operations.createTopic(createTopic);
 
             // Quick n dirty json response
-            return new ResultResponse("CreateTopic", result, "");
+            return new ResultResponse("CreateTopic", result, "Created topic '" + createTopicRequest.getName() + "'");
         } catch (final Exception e) {
             throw new ApiException("CreateTopic", e);
         }
@@ -386,6 +397,34 @@ public List<ConfigItem> modifyTopicConfig(
         }
     }
 
+    /**
+     * POST Delete existing topic on cluster.
+     * This should require ADMIN role.
+     *
+     * @TODO explicitly disabled until custom user roles. https://github.com/SourceLabOrg/kafka-webview/issues/157
+     */
+//    @ResponseBody
+//    @RequestMapping(path = "/cluster/{id}/delete/topic", method = RequestMethod.POST, produces = "application/json")
+    public ResultResponse deleteTopic(@PathVariable final Long id, @RequestBody final DeleteTopicRequest deleteTopicRequest) {
+        // Retrieve cluster
+        final Cluster cluster = retrieveClusterById(id);
+
+        final String name = deleteTopicRequest.getName();
+        if (name == null || name.trim().isEmpty()) {
+            throw new ApiException("DeleteTopic", "Invalid topic name");
+        }
+
+        // Create new Operational Client
+        try (final KafkaOperations operations = createOperationsClient(cluster)) {
+            final boolean result = operations.removeTopic(deleteTopicRequest.getName());
+
+            // Quick n dirty json response
+            return new ResultResponse("DeleteTopic", result, "Removed topic '" + deleteTopicRequest.getName() + "'");
+        } catch (final Exception e) {
+            throw new ApiException("DeleteTopic", e);
+        }
+    }
+
     /**
      * GET Nodes within a cluster.
      */

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/requests/DeleteTopicRequest.java

@@ -0,0 +1,47 @@
+/**
+ * MIT License
+ *
+ * Copyright (c) 2017, 2018, 2019 SourceLab.org (https://github.com/SourceLabOrg/kafka-webview/)
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.sourcelab.kafka.webview.ui.controller.api.requests;
+
+/**
+ * Represents a request to remove an existing topic from kafka cluster.
+ */
+public class DeleteTopicRequest {
+    private String name;
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(final String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String toString() {
+        return "DeleteTopicRequest{"
+            + "name='" + name + '\''
+            + '}';
+    }
+}

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperations.java

@@ -30,6 +30,8 @@
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.DeleteConsumerGroupsResult;
+import org.apache.kafka.clients.admin.DeleteTopicsOptions;
+import org.apache.kafka.clients.admin.DeleteTopicsResult;
 import org.apache.kafka.clients.admin.DescribeConfigsResult;
 import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult;
 import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
@@ -291,6 +293,26 @@ public TopicConfig alterTopicConfig(final String topic, final Map<String, String
         }
     }
 
+    /**
+     * Remove a topic from a kafka cluster.  This is destructive!
+     * @param topic name of the topic to remove.
+     * @return true on success.
+     */
+    public boolean removeTopic(final String topic) {
+        try {
+            final DeleteTopicsResult result = adminClient.deleteTopics(Collections.singletonList(topic));
+
+            // Wait for the async request to process.
+            result.all().get();
+
+            // return true?
+            return true;
+        } catch (final InterruptedException | ExecutionException exception) {
+            // TODO Handle this
+            throw new RuntimeException(exception.getMessage(), exception);
+        }
+    }
+
     /**
      * List all consumer groups on server.
      * @return Immutable sorted list of Consumer Group Identifiers.

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/TopicList.java

@@ -64,6 +64,28 @@ public List<String> getTopicNames() {
         return Collections.unmodifiableList(topicNames);
     }
 
+    /**
+     * Given a search value for topic names, return all entries that match search.
+     * @param search search string.
+     * @return filtered TopicList.
+     */
+    public TopicList filterByTopicName(final String search) {
+        // Null means match nothing.
+        if (search == null) {
+            return new TopicList(Collections.emptyList());
+        }
+        final String normalizedSearch = search.toLowerCase();
+
+        final List<TopicListing> topicListings = new ArrayList<>();
+        for (final TopicListing topicListing : getTopics()) {
+            if (topicListing.getName().toLowerCase().contains(normalizedSearch)) {
+                topicListings.add(topicListing);
+            }
+        }
+
+        return new TopicList(topicListings);
+    }
+
     @Override
     public String toString() {
         return "TopicList{"