Add ability to modify a topics configuration via the UI. (#95)

Crim · web-flow · commit a3620d1ca340 · 2018-09-23T13:32:59.000+09:00
* Add ability to modify configuration for topics

* fix security requirements

* tie in UI to be able to modify topic configs

* restrict UI if not authorized
diff --git a/kafka-webview-ui/src/main/frontend/js/app.js b/kafka-webview-ui/src/main/frontend/js/app.js
@@ -273,6 +273,27 @@ var ApiClient = {
             }
         });
     },
+    modifyTopicConfig: function(clusterId, topic, configName, configValue, callback) {
+        var payloadJson = {
+            topic: topic,
+            config: {}
+        };
+        payloadJson['config'][configName] = configValue;
+
+        var payload = JSON.stringify(payloadJson);
+        jQuery.ajax({
+            type: 'POST',
+            url: '/api/cluster/' + clusterId + '/modify/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');
+            }
+        });
+    },
     defaultErrorHandler: function(jqXHR, textStatus, errorThrown) {
         // convert response to json
         var response = jQuery.parseJSON(jqXHR.responseText);
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/configuration/SecurityConfig.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/configuration/SecurityConfig.java
@@ -119,7 +119,10 @@ private void enableUserAuth(final HttpSecurity http) throws Exception {
                 "/configuration/**",
 
                 // Create topic
-                "/api/cluster/*/create/**"
+                "/api/cluster/*/create/**",
+
+                // Modify topic
+                "/api/cluster/*/modify/**"
             ).hasRole("ADMIN")
 
             // All other requests must be authenticated
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/ApiController.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/ApiController.java
@@ -25,6 +25,12 @@
 package org.sourcelab.kafka.webview.ui.controller.api;
 
 import org.sourcelab.kafka.webview.ui.controller.BaseController;
+import org.sourcelab.kafka.webview.ui.controller.api.exceptions.ApiException;
+import org.sourcelab.kafka.webview.ui.controller.api.exceptions.NotFoundApiException;
+import org.sourcelab.kafka.webview.ui.controller.api.requests.ConsumeRequest;
+import org.sourcelab.kafka.webview.ui.controller.api.requests.CreateTopicRequest;
+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;
 import org.sourcelab.kafka.webview.ui.manager.kafka.KafkaOperationsFactory;
 import org.sourcelab.kafka.webview.ui.manager.kafka.SessionIdentifier;
@@ -330,6 +336,37 @@ public ResultResponse createTopic(@PathVariable final Long id, @RequestBody fina
         }
     }
 
+    /**
+     * POST Modify a topic's configuration on cluster.
+     * This should require ADMIN role.
+     */
+    @ResponseBody
+    @RequestMapping(path = "/cluster/{id}/modify/topic", method = RequestMethod.POST, produces = "application/json")
+    public List<ConfigItem> modifyTopicConfig(
+        @PathVariable final Long id,
+        @RequestBody final ModifyTopicConfigRequest modifyTopicConfigRequest
+    ) {
+        // Retrieve cluster
+        final Cluster cluster = retrieveClusterById(id);
+
+        final String name = modifyTopicConfigRequest.getTopic();
+        if (name == null || name.trim().isEmpty()) {
+            throw new ApiException("ModifyTopic", "Invalid topic name");
+        }
+
+        final Map<String, String> configEntries = modifyTopicConfigRequest.getConfig();
+        if (configEntries == null || configEntries.isEmpty()) {
+            throw new ApiException("ModifyTopic", "Invalid configuration defined");
+        }
+
+        // Create new Operational Client
+        try (final KafkaOperations operations = createOperationsClient(cluster)) {
+            return operations.alterTopicConfig(name, configEntries).getConfigEntries();
+        } catch (final Exception e) {
+            throw new ApiException("ModifyTopic", e);
+        }
+    }
+
     /**
      * GET Nodes within a cluster.
      */
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/exceptions/ApiException.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/exceptions/ApiException.java
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package org.sourcelab.kafka.webview.ui.controller.api;
+package org.sourcelab.kafka.webview.ui.controller.api.exceptions;
 
 /**
  * Represents an error that occurred within an API request.
@@ -41,7 +41,7 @@ public ApiException(final String type, final Throwable cause) {
     /**
      * Constructor.
      */
-    ApiException(final String type, final String message) {
+    public ApiException(final String type, final String message) {
         super(message);
         this.type = type;
     }
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/exceptions/NotFoundApiException.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/exceptions/NotFoundApiException.java
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package org.sourcelab.kafka.webview.ui.controller.api;
+package org.sourcelab.kafka.webview.ui.controller.api.exceptions;
 
 /**
  * Error when requesting an object over the API that doesn't not exist.
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/requests/ConsumeRequest.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/requests/ConsumeRequest.java
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package org.sourcelab.kafka.webview.ui.controller.api;
+package org.sourcelab.kafka.webview.ui.controller.api.requests;
 
 import java.util.List;
 import java.util.Map;
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/requests/CreateTopicRequest.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/requests/CreateTopicRequest.java
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package org.sourcelab.kafka.webview.ui.controller.api;
+package org.sourcelab.kafka.webview.ui.controller.api.requests;
 
 /**
  * Represents a request to create a new topic.
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/requests/ModifyTopicConfigRequest.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/requests/ModifyTopicConfigRequest.java
@@ -0,0 +1,51 @@
+/**
+ * MIT License
+ *
+ * Copyright (c) 2017, 2018 SourceLab.org (https://github.com/Crim/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;
+
+import java.util.Map;
+
+/**
+ * Represents a request to modify a topic configuration.
+ */
+public class ModifyTopicConfigRequest {
+    private String topic;
+    private Map<String, String> config;
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public Map<String, String> getConfig() {
+        return config;
+    }
+
+    @Override
+    public String toString() {
+        return "ModifyTopicConfigRequest{"
+            + "topic='" + topic + '\''
+            + ", config=" + config
+            + '}';
+    }
+}
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/responses/ResultResponse.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/api/responses/ResultResponse.java
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package org.sourcelab.kafka.webview.ui.controller.api;
+package org.sourcelab.kafka.webview.ui.controller.api.responses;
 
 /**
  * Generic response object.
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/stream/StreamController.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/stream/StreamController.java
@@ -25,7 +25,7 @@
 package org.sourcelab.kafka.webview.ui.controller.stream;
 
 import org.sourcelab.kafka.webview.ui.controller.BaseController;
-import org.sourcelab.kafka.webview.ui.controller.api.ConsumeRequest;
+import org.sourcelab.kafka.webview.ui.controller.api.requests.ConsumeRequest;
 import org.sourcelab.kafka.webview.ui.manager.kafka.SessionIdentifier;
 import org.sourcelab.kafka.webview.ui.manager.kafka.ViewCustomizer;
 import org.sourcelab.kafka.webview.ui.manager.kafka.config.FilterDefinition;
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperations.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperations.java
@@ -25,6 +25,7 @@
 package org.sourcelab.kafka.webview.ui.manager.kafka;
 
 import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AlterConfigsResult;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
@@ -226,6 +227,41 @@ public boolean createTopic(final CreateTopic createTopic) {
         }
     }
 
+    /**
+     * Modify configuration values for a specific topic.
+     * @param topic The topic to modify.
+     * @param configItems Map of Key => Value to modify.
+     * @return boolean
+     */
+    public TopicConfig alterTopicConfig(final String topic, final Map<String, String> configItems) {
+        try {
+            // Define the resource we want to modify, the topic.
+            final ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
+
+            final List<ConfigEntry> configEntries = new ArrayList<>();
+            for (final Map.Entry<String, String> entry : configItems.entrySet()) {
+                configEntries.add(
+                    new ConfigEntry(entry.getKey(), entry.getValue())
+                );
+            }
+
+            // Define the configuration set
+            final Config config = new Config(configEntries);
+
+            // Create the topic
+            final AlterConfigsResult result = adminClient.alterConfigs(Collections.singletonMap(configResource, config));
+
+            // Wait for the async request to process.
+            result.all().get();
+
+            // Lets return updated topic details
+            return getTopicConfig(topic);
+        } catch (final InterruptedException | ExecutionException exception) {
+            // TODO Handle this
+            throw new RuntimeException(exception.getMessage(), exception);
+        }
+    }
+
     private List<ConfigItem> describeResource(final ConfigResource configResource) {
         final DescribeConfigsResult result = adminClient.describeConfigs(Collections.singleton(configResource));
 
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/ViewCustomizer.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/ViewCustomizer.java
@@ -24,7 +24,7 @@
 
 package org.sourcelab.kafka.webview.ui.manager.kafka;
 
-import org.sourcelab.kafka.webview.ui.controller.api.ConsumeRequest;
+import org.sourcelab.kafka.webview.ui.controller.api.requests.ConsumeRequest;
 import org.sourcelab.kafka.webview.ui.manager.kafka.config.FilterDefinition;
 import org.sourcelab.kafka.webview.ui.manager.socket.StartingPosition;
 import org.sourcelab.kafka.webview.ui.model.Filter;
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/BrokerConfig.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/BrokerConfig.java
@@ -35,6 +35,10 @@
 public class BrokerConfig {
     private final List<ConfigItem> configEntries;
 
+    /**
+     * Constructor.
+     * @param configEntries entries in the config.
+     */
     public BrokerConfig(final List<ConfigItem> configEntries) {
         // Create new list from source.
         final List<ConfigItem> sorted = new ArrayList<>(configEntries);
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/PartitionDetails.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/PartitionDetails.java
@@ -36,6 +36,7 @@ public class PartitionDetails {
     private final NodeDetails leader;
     private final List<NodeDetails> replicas;
     private final List<NodeDetails> isr;
+    private final boolean isUnderReplicated;
 
     /**
      * Constructor.
@@ -52,6 +53,10 @@ public PartitionDetails(
         this.leader = leader;
         this.replicas = Collections.unmodifiableList(replicas);
         this.isr = Collections.unmodifiableList(isr);
+
+        // Lazy many way to determine this is to count the number of replicas
+        // and compare it to the count of ISR, if not equal, then under replicated.
+        this.isUnderReplicated = getReplicas().size() != getIsr().size();
     }
 
     public String getTopic() {
@@ -74,6 +79,10 @@ public List<NodeDetails> getIsr() {
         return isr;
     }
 
+    public boolean isUnderReplicated() {
+        return isUnderReplicated;
+    }
+
     @Override
     public String toString() {
         return "PartitionDetails{"
@@ -82,6 +91,7 @@ public String toString() {
             + ", leader=" + leader
             + ", replicas=" + replicas
             + ", isr=" + isr
+            + ", isUnderReplicated=" + isUnderReplicated
             + '}';
     }
 }
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/TopicConfig.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/TopicConfig.java
@@ -28,11 +28,15 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Optional;
 
 /**
  * Represents a collection of metadata about how a topic is configured.
  */
 public class TopicConfig {
+    /**
+     * It may make more sense to make this into a map for faster direct access by name in the future.
+     */
     private final List<ConfigItem> configEntries;
 
     /**
@@ -53,6 +57,18 @@ public List<ConfigItem> getConfigEntries() {
         return configEntries;
     }
 
+    /**
+     * Given a configuration name, return its entry.
+     * @param name name of config value to find.
+     * @return Optionally, the matching ConfigItem.
+     */
+    public Optional<ConfigItem> getConfigItemByName(final String name) {
+        return getConfigEntries()
+            .stream()
+            .filter((entry) -> name.equals(entry.getName()))
+            .findFirst();
+    }
+
     @Override
     public String toString() {
         return "TopicConfig{"
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/TopicDetails.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/TopicDetails.java
@@ -34,6 +34,7 @@ public class TopicDetails {
     private final String name;
     private final boolean isInternal;
     private final List<PartitionDetails> partitions;
+    private final boolean isUnderReplicated;
 
     /**
      * Constructor.
@@ -42,6 +43,16 @@ public TopicDetails(final String name, final boolean isInternal, final List<Part
         this.name = name;
         this.isInternal = isInternal;
         this.partitions = Collections.unmodifiableList(partitions);
+
+        // Determine if any partition is under replicated.
+        boolean underReplicated = false;
+        for (final PartitionDetails partitionDetails : partitions) {
+            if (partitionDetails.isUnderReplicated()) {
+                underReplicated = true;
+                break;
+            }
+        }
+        this.isUnderReplicated = underReplicated;
     }
 
     public String getName() {
@@ -56,12 +67,20 @@ public boolean isInternal() {
         return isInternal;
     }
 
+    /**
+     * @return true if any partition on the topic is under replicated.
+     */
+    public boolean isUnderReplicated() {
+        return isUnderReplicated;
+    }
+
     @Override
     public String toString() {
         return "TopicDetails{"
-            + "+ name='" + name + '\''
-            + ", + isInternal=" + isInternal
-            + ", + partitions=" + partitions
+            + "name='" + name + '\''
+            + ", isInternal=" + isInternal
+            + ", partitions=" + partitions
+            + ", isUnderReplicated=" + isUnderReplicated
             + '}';
     }
 }
diff --git a/kafka-webview-ui/src/main/resources/templates/cluster/read.html b/kafka-webview-ui/src/main/resources/templates/cluster/read.html
diff --git a/kafka-webview-ui/src/main/resources/templates/cluster/readTopic.html b/kafka-webview-ui/src/main/resources/templates/cluster/readTopic.html
diff --git a/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/AbstractMvcTest.java b/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/AbstractMvcTest.java
diff --git a/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/api/ApiControllerTest.java b/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/api/ApiControllerTest.java
diff --git a/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperationsTest.java b/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperationsTest.java
diff --git a/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/ViewCustomizerTest.java b/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/ViewCustomizerTest.java
