Issue 462 - Unsubscribe topics from consumer group

api/src/main/java/io/kafbat/ui/controller/ConsumerGroupsController.java

@@ -1,6 +1,7 @@
 package io.kafbat.ui.controller;
 
 import static io.kafbat.ui.model.rbac.permission.ConsumerGroupAction.DELETE;
+import static io.kafbat.ui.model.rbac.permission.ConsumerGroupAction.DELETE_OFFSETS;
 import static io.kafbat.ui.model.rbac.permission.ConsumerGroupAction.RESET_OFFSETS;
 import static io.kafbat.ui.model.rbac.permission.ConsumerGroupAction.VIEW;
 import static java.util.stream.Collectors.toMap;
@@ -188,6 +189,24 @@ public Mono<ResponseEntity<Void>> resetConsumerGroupOffsets(String clusterName,
     }).thenReturn(ResponseEntity.ok().build());
   }
 
+  @Override
+  public Mono<ResponseEntity<Void>> deleteConsumerGroupOffsets(String clusterName,
+                                                              String group,
+                                                              String topic,
+                                                              ServerWebExchange exchange) {
+    var context = AccessContext.builder()
+        .cluster(clusterName)
+        .topicActions(topic, TopicAction.VIEW)
+        .consumerGroupActions(group, DELETE_OFFSETS)
+        .operationName("resetConsumerGroupOffsets")
+        .build();
+
+    return validateAccess(context)
+        .then(consumerGroupService.deleteConsumerGroupOffsets(getCluster(clusterName), group, topic))
+        .doOnEach(sig -> audit(context, sig))
+        .thenReturn(ResponseEntity.ok().build());
+  }
+
   private ConsumerGroupsPageResponseDTO convertPage(ConsumerGroupService.ConsumerGroupsPage
                                                         consumerGroupConsumerGroupsPage) {
     return new ConsumerGroupsPageResponseDTO()

api/src/main/java/io/kafbat/ui/model/rbac/permission/ConsumerGroupAction.java

@@ -8,11 +8,10 @@ public enum ConsumerGroupAction implements PermissibleAction {
 
   VIEW,
   DELETE(VIEW),
-  RESET_OFFSETS(VIEW)
+  RESET_OFFSETS(VIEW),
+  DELETE_OFFSETS(VIEW);
 
-  ;
-
-  public static final Set<ConsumerGroupAction> ALTER_ACTIONS = Set.of(DELETE, RESET_OFFSETS);
+  public static final Set<ConsumerGroupAction> ALTER_ACTIONS = Set.of(DELETE, RESET_OFFSETS, DELETE_OFFSETS);
 
   private final ConsumerGroupAction[] dependantActions;
 

api/src/main/java/io/kafbat/ui/service/ConsumerGroupService.java

@@ -270,4 +270,11 @@ public EnhancedConsumer createConsumer(KafkaCluster cluster,
     );
   }
 
+  public Mono<Void> deleteConsumerGroupOffsets(KafkaCluster cluster,
+                                               String groupId,
+                                               String topic) {
+    return adminClientService
+        .get(cluster)
+        .flatMap(adminClient -> adminClient.deleteConsumerGroupOffsets(groupId, topic));
+  }
 }

api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java

@@ -531,6 +531,21 @@ public Mono<Void> alterConsumerGroupOffsets(String groupId, Map<TopicPartition,
         .all());
   }
 
+  public Mono<Void> deleteConsumerGroupOffsets(String groupId, String topic) {
+    var offsets = listConsumerGroupOffsets(List.of(groupId), null);
+    Mono<Set<TopicPartition>> partitions = offsets.map(tps ->
+        tps.row(groupId)
+            .keySet()
+            .stream()
+            .filter(tp -> tp.topic().equals(topic))
+            .collect(Collectors.toSet())
+    );
+
+    return partitions.flatMap(partitionSet ->
+        toMono(client.deleteConsumerGroupOffsets(groupId, partitionSet).all())
+    );
+  }
+
   /**
    * List offset for the topic's partitions and OffsetSpec.
    *

contract/src/main/resources/swagger/kafbat-ui-api.yaml

@@ -1048,6 +1048,33 @@ paths:
         200:
           description: OK
 
+  /api/clusters/{clusterName}/consumer-groups/{id}/topics/{topic}:
+    delete:
+      tags:
+        - Consumer Groups
+      summary: deletes consumer group offsets
+      operationId: deleteConsumerGroupOffsets
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: id
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: topic
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        200:
+          description: OK
+
+
   /api/clusters/{clusterName}/schemas:
     post:
       tags:
@@ -3828,6 +3855,7 @@ components:
         - CREATE
         - DELETE
         - RESET_OFFSETS
+        - DELETE_OFFSETS
         - EXECUTE
         - MODIFY_GLOBAL_COMPATIBILITY
         - ANALYSIS_VIEW

frontend/src/components/ConsumerGroups/Details/ListItem.tsx

@@ -1,14 +1,18 @@
 import React from 'react';
-import { ConsumerGroupTopicPartition } from 'generated-sources';
+import {Action, ConsumerGroupTopicPartition, ResourceType} from 'generated-sources';
 import { Link } from 'react-router-dom';
 import { ClusterName } from 'lib/interfaces/cluster';
-import { clusterTopicPath } from 'lib/paths';
+import {ClusterGroupParam, clusterTopicPath} from 'lib/paths';
 import MessageToggleIcon from 'components/common/Icons/MessageToggleIcon';
 import IconButtonWrapper from 'components/common/Icons/IconButtonWrapper';
 import { TableKeyLink } from 'components/common/table/Table/TableKeyLink.styled';
 
 import TopicContents from './TopicContents/TopicContents';
 import { FlexWrapper } from './ListItem.styled';
+import {Dropdown} from "../../common/Dropdown";
+import {ActionDropdownItem} from "../../common/ActionComponent";
+import useAppParams from "../../../lib/hooks/useAppParams";
+import {useDeleteConsumerGroupOffsetsMutation} from "../../../lib/hooks/api/consumers";
 
 interface Props {
   clusterName: ClusterName;
@@ -18,10 +22,12 @@
 
 const ListItem: React.FC<Props> = ({ clusterName, name, consumers }) => {
   const [isOpen, setIsOpen] = React.useState(false);
+  const consumer = useAppParams<ClusterGroupParam>()
+  const deleteOffset = useDeleteConsumerGroupOffsetsMutation(consumer)
 
   const getTotalconsumerLag = () => {
     let count = 0;
     consumers.forEach((consumer) => {
       count += consumer?.consumerLag || 0;
     });
     return count;
@@ -40,7 +46,22 @@
             </TableKeyLink>
           </FlexWrapper>
         </td>
-        <td>{getTotalconsumerLag()}</td>
+        <td>
+          {getTotalconsumerLag()}
+        </td>
+        <td>
+          <Dropdown>
+            <ActionDropdownItem
+              onClick={() => deleteOffset.mutateAsync(name)}
+              permission={{
+                resource: ResourceType.CONSUMER,
+                action: Action.DELETE_OFFSETS,
+                value: consumer.consumerGroupID
+              }}>
+              Unsubscribe from topic
+            </ActionDropdownItem>
+          </Dropdown>
+        </td>
       </tr>
       {isOpen && <TopicContents consumers={consumers} />}
     </>

frontend/src/lib/hooks/api/consumers.ts

@@ -90,3 +90,30 @@ export const useResetConsumerGroupOffsetsMutation = ({
     }
   );
 };
+
+export const useDeleteConsumerGroupOffsetsMutation = ({
+                                                       clusterName,
+                                                       consumerGroupID,
+                                                     }: UseConsumerGroupDetailsProps) => {
+  const queryClient = useQueryClient();
+  return useMutation(
+    (topic: string) =>
+      api.deleteConsumerGroupOffsets({
+        clusterName,
+        id: consumerGroupID,
+        topic,
+      }),
+    {
+      onSuccess: () => {
+        showSuccessAlert({
+          message: `Consumer ${consumerGroupID} group offsets deleted.`,
+        });
+        queryClient.invalidateQueries([
+          'clusters',
+          clusterName,
+          'consumerGroups',
+        ]);
+      },
+    }
+  );
+};