content: package io.aiven.klaw.clusterapi.controller;
  
  import io.aiven.klaw.clusterapi.models.ApiResponse;
  import io.aiven.klaw.clusterapi.models.consumergroup.OffsetDetails;
  import io.aiven.klaw.clusterapi.models.consumergroup.ResetConsumerGroupOffsetsRequest;
  import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
  import io.aiven.klaw.clusterapi.services.ConsumerGroupService;
  import jakarta.validation.Valid;
  import java.util.List;
  import lombok.AllArgsConstructor;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.http.HttpStatus;
  import org.springframework.http.MediaType;
  import org.springframework.http.ResponseEntity;
  import org.springframework.web.bind.annotation.PathVariable;
  import org.springframework.web.bind.annotation.PostMapping;
  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.RestController;
  
  @RestController
  @RequestMapping("/topics")
  @Slf4j
  @AllArgsConstructor
  public class ConsumerGroupsController {
    ConsumerGroupService consumerGroupService;
  
    /**
     * Reset offsets of a consumer group provided topic name, consumer group, reset type(earliest,
     * latest), and timestamp if applicable
     *
     * @param bootstrapServers environment param
     * @param protocol SSL or plaintext
     * @param clusterIdentification cluster id
     * @param consumerGroupOffsetsRequest request object containing topic, group and reset type
     * @return ApiResponse with OffsetReset data object
     * @throws Exception any exception while retrieving or resetting offsets
     */
    @PostMapping(
        value = "/consumerGroupOffsets/reset/{bootstrapServers}/{protocol}/{clusterIdentification}",
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ApiResponse resetConsumerOffsets(
        @PathVariable String bootstrapServers,
        @Valid @PathVariable KafkaSupportedProtocol protocol,
        @PathVariable String clusterIdentification,
        @RequestBody @Valid ResetConsumerGroupOffsetsRequest consumerGroupOffsetsRequest)
        throws Exception {
      return consumerGroupService.resetConsumerGroupOffsets(
          bootstrapServers, protocol, clusterIdentification, consumerGroupOffsetsRequest);
    }
  
    /**
     * @param bootstrapServers env param
     * @param protocol ssl or plaintext protocol
     * @param clusterName cluster id
     * @param consumerGroupId consumer group
     * @param topicName topic name
     * @return current offset positions
     * @throws Exception any exception while retrieving offsets
     */
    @RequestMapping(
        value =
            "/consumerGroupOffsets/{bootstrapServers}/{protocol}/{clusterName}/{consumerGroupId}/{topicName}",
        method = RequestMethod.GET,
        produces = {MediaType.APPLICATION_JSON_VALUE})
    public ResponseEntity<List<OffsetDetails>> getConsumerOffsets(
        @PathVariable String bootstrapServers,
        @Valid @PathVariable KafkaSupportedProtocol protocol,
        @PathVariable String clusterName,
        @PathVariable String consumerGroupId,
        @PathVariable String topicName)
        throws Exception {
      List<OffsetDetails> consumerOffsetDetails =
          consumerGroupService.getConsumerGroupDetails(
              consumerGroupId, topicName, bootstrapServers, protocol, clusterName);
  
      return new ResponseEntity<>(consumerOffsetDetails, HttpStatus.OK);
    }
  }
  