Allow to delete consumer group

sajiljosephs · Sep 28, 2018 · ef77f05 · ef77f05
1 parent 2b0f284
commit ef77f05
Show file tree

Hide file tree

Showing 4 changed files with 46 additions and 1 deletion.
diff --git a/public/includes/group.ftl b/public/includes/group.ftl
@@ -41,6 +41,12 @@
                             <td class="row-action main-row-action">
                                 <a href="/${clusterId}/group/${group.getId()}" ><i class="fa fa-search"></i></a>
                             </td>
+                            <td class="row-action">
+                                <a
+                                    href="/${clusterId}/group/${group.getId()}/delete"
+                                    data-confirm="Do you want to delete consumer group <br /><strong>${group.getId()}</strong><br /><br /> ?"
+                                ><i class="fa fa-trash"></i></a>
+                            </td>
                         </tr>
                     </#list>
             </tbody>

diff --git a/src/main/java/org/kafkahq/controllers/GroupController.java b/src/main/java/org/kafkahq/controllers/GroupController.java
@@ -2,17 +2,23 @@
 
 import com.google.inject.Inject;
 import org.jooby.Request;
+import org.jooby.Result;
 import org.jooby.Results;
 import org.jooby.View;
 import org.jooby.mvc.GET;
 import org.jooby.mvc.Path;
 import org.kafkahq.models.ConsumerGroup;
 import org.kafkahq.repositories.ConsumerGroupRepository;
+import org.kafkahq.response.ResultStatusResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ExecutionException;
 
 @Path("/{cluster}/group")
 public class GroupController extends AbstractController {
+    private static final Logger logger = LoggerFactory.getLogger(TopicController.class);
+
     @Inject
     private ConsumerGroupRepository consumerGroupRepository;
 
@@ -50,4 +56,27 @@ public View group(Request request, String tab) throws ExecutionException, Interr
         );
     }
 
+
+    @GET
+    @Path("{id}/delete")
+    public Result delete(Request request) {
+        String name = request.param("id").value();
+        ResultStatusResponse result = new ResultStatusResponse();
+
+        try {
+            this.consumerGroupRepository.delete(request.param("cluster").value(), name);
+
+            result.result = true;
+            result.message = "Topic '" + name + "' is deleted";
+
+            return Results.with(result, 200);
+        } catch (Exception exception) {
+            logger.error("Failed to delete topic " + name, exception);
+
+            result.result = false;
+            result.message = exception.getCause().getMessage();
+
+            return Results.with(result, 500);
+        }
+    }
 }
diff --git a/src/main/java/org/kafkahq/repositories/ConsumerGroupRepository.java b/src/main/java/org/kafkahq/repositories/ConsumerGroupRepository.java
@@ -1,6 +1,7 @@
 package org.kafkahq.repositories;
 
 import com.google.inject.Binder;
+import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.typesafe.config.Config;
 import org.apache.kafka.clients.admin.ConsumerGroupDescription;
@@ -11,13 +12,17 @@
 import org.jooby.Jooby;
 import org.kafkahq.models.ConsumerGroup;
 import org.kafkahq.models.Partition;
+import org.kafkahq.modules.KafkaModule;
 
 import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
 @Singleton
 public class ConsumerGroupRepository extends AbstractRepository implements Jooby.Module {
+    @Inject
+    private KafkaModule kafkaModule;
+
     public List<ConsumerGroup> list() throws ExecutionException, InterruptedException {
         ArrayList<String> list = new ArrayList<>();
 
@@ -76,6 +81,12 @@ public List<ConsumerGroup> findByTopic(String topic) throws ExecutionException,
             .collect(Collectors.toList());
     }
 
+    public void delete(String clusterId, String name) throws ExecutionException, InterruptedException {
+        kafkaModule.getAdminClient(clusterId).deleteConsumerGroups(new ArrayList<String>() {{
+            add(name);
+        }}).all().get();
+    }
+
     @SuppressWarnings("NullableProblems")
     @Override
     public void configure(Env env, Config conf, Binder binder) {

diff --git a/src/main/java/org/kafkahq/repositories/TopicRepository.java b/src/main/java/org/kafkahq/repositories/TopicRepository.java
@@ -75,7 +75,6 @@ public void delete(String clusterId, String name) throws ExecutionException, Int
         }}).all().get();
     }
 
-
     @SuppressWarnings("NullableProblems")
     @Override
     public void configure(Env env, Config conf, Binder binder) {