feat: implement deleteConsumerGroupOffsets

Author: p-eye

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

@@ -16,12 +16,14 @@
 import io.kafbat.ui.model.PartitionOffsetDTO;
 import io.kafbat.ui.model.SortOrderDTO;
 import io.kafbat.ui.model.rbac.AccessContext;
+import io.kafbat.ui.model.rbac.permission.ConnectAction;
 import io.kafbat.ui.model.rbac.permission.TopicAction;
 import io.kafbat.ui.service.ConsumerGroupService;
 import io.kafbat.ui.service.OffsetsResetService;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
@@ -59,6 +61,22 @@ public Mono<ResponseEntity<Void>> deleteConsumerGroup(String clusterName,
         .thenReturn(ResponseEntity.ok().build());
   }
 
+  @Override
+  public Mono<ResponseEntity<Void>> deleteConsumerGroupOffsets(String clusterName, String groupId, String topicName,
+                                                               ServerWebExchange exchange) {
+    var context = AccessContext.builder()
+        .cluster(clusterName)
+        .consumerGroupActions(groupId, RESET_OFFSETS)
+        .topicActions(topicName, TopicAction.VIEW)
+        .operationName("deleteConsumerGroupOffsets")
+        .build();
+
+    return validateAccess(context)
+        .then(consumerGroupService.deleteConsumerGroupOffset(getCluster(clusterName), groupId, topicName))
+        .doOnEach(sig -> audit(context, sig))
+        .thenReturn(ResponseEntity.ok().build());
+  }
+
   @Override
   public Mono<ResponseEntity<ConsumerGroupDetailsDTO>> getConsumerGroup(String clusterName,
                                                                         String consumerGroupId,

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

@@ -18,6 +18,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 import java.util.function.ToIntFunction;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -247,6 +248,13 @@ public Mono<Void> deleteConsumerGroupById(KafkaCluster cluster,
         .flatMap(adminClient -> adminClient.deleteConsumerGroups(List.of(groupId)));
   }
 
+  public Mono<Void> deleteConsumerGroupOffset(KafkaCluster cluster,
+                                            String groupId,
+                                              String topicName) {
+    return adminClientService.get(cluster)
+        .flatMap(adminClient -> adminClient.deleteConsumerGroupOffsets(groupId, topicName));
+  }
+
   public EnhancedConsumer createConsumer(KafkaCluster cluster) {
     return createConsumer(cluster, Map.of());
   }

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

@@ -49,6 +49,7 @@
 import org.apache.kafka.clients.admin.DescribeClusterOptions;
 import org.apache.kafka.clients.admin.DescribeClusterResult;
 import org.apache.kafka.clients.admin.DescribeConfigsOptions;
+import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions;
 import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
 import org.apache.kafka.clients.admin.ListOffsetsResult;
 import org.apache.kafka.clients.admin.ListTopicsOptions;
@@ -74,6 +75,7 @@
 import org.apache.kafka.common.errors.ClusterAuthorizationException;
 import org.apache.kafka.common.errors.GroupIdNotFoundException;
 import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.errors.GroupSubscribedToTopicException;
 import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.errors.SecurityDisabledException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
@@ -436,6 +438,23 @@ public Mono<Void> deleteConsumerGroups(Collection<String> groupIds) {
             th -> Mono.error(new IllegalEntityStateException("The group is not empty")));
   }
 
+  public Mono<Void> deleteConsumerGroupOffsets(String groupId, String topicName) {
+    return listConsumerGroupOffsets(List.of(groupId), null)
+        .flatMap(table -> {
+          // filter TopicPartitions by topicName
+          Set<TopicPartition> partitions = table.row(groupId).keySet().stream()
+              .filter(tp -> tp.topic().equals(topicName))
+              .collect(Collectors.toSet());
+          return toMono(client.deleteConsumerGroupOffsets(groupId, partitions).all());
+        })
+        .onErrorResume(GroupIdNotFoundException.class,
+            th -> Mono.error(new NotFoundException("The group id does not exist")))
+        .onErrorResume(UnknownTopicOrPartitionException.class,
+            th -> Mono.error(new NotFoundException("The topic or partition is unknown")))
+        .onErrorResume(GroupSubscribedToTopicException.class,
+            th -> Mono.error(new IllegalEntityStateException("The group is not empty")));
+  }
+
   public Mono<Void> createTopic(String name,
                                 int numPartitions,
                                 @Nullable Integer replicationFactor,