package io.aiven.klaw.clusterapi;

import static io.aiven.klaw.clusterapi.models.enums.ClusterStatus.ONLINE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.when;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.aiven.klaw.clusterapi.models.ApiResponse;
import io.aiven.klaw.clusterapi.models.ClusterAclRequest;
import io.aiven.klaw.clusterapi.models.ClusterTopicRequest;
import io.aiven.klaw.clusterapi.models.consumergroup.OffsetResetType;
import io.aiven.klaw.clusterapi.models.consumergroup.OffsetsTiming;
import io.aiven.klaw.clusterapi.models.consumergroup.ResetConsumerGroupOffsetsRequest;
import io.aiven.klaw.clusterapi.models.enums.AclIPPrincipleType;
import io.aiven.klaw.clusterapi.models.enums.AclType;
import io.aiven.klaw.clusterapi.models.enums.AclsNativeType;
import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
import io.aiven.klaw.clusterapi.models.enums.ClusterStatus;
import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
import io.aiven.klaw.clusterapi.models.enums.RequestOperationType;
import io.aiven.klaw.clusterapi.services.SchemaService;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.acl.AccessControlEntryFilter;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.resource.ResourceType;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.http.MediaType;
import org.springframework.kafka.test.EmbeddedKafkaZKBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.mock.web.MockHttpServletResponse;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;

@SpringBootTest(
    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
    classes = KafkaClusterApiApplication.class)
@AutoConfigureMockMvc
@TestPropertySource(locations = "classpath:application.properties")
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@DirtiesContext
@EmbeddedKafka(
    brokerProperties = {
      "authorizer.class.name=kafka.security.authorizer.AclAuthorizer",
      "allow.everyone.if.no.acl.found=true",
      "super.users=User:ANONYMOUS",
      "ssl.truststore.location=src/test/resources/selfsignedcerts/truststore.jks",
      "ssl.truststore.password=klaw1234",
      "ssl.keystore.location=src/test/resources/selfsignedcerts/keystore.p12",
      "ssl.key.password=klaw1234",
      "ssl.keystore.password=klaw1234",
      "ssl.keystore.type=pkcs12",
      "listeners=PLAINTEXT://"
          + ClusterApiControllerIT.BOOTSTRAP_SERVERS
          + ",SSL://"
          + ClusterApiControllerIT.BOOTSTRAP_SERVERS_SSL
    },
    partitions = 1,
    adminTimeout = 100)
@Slf4j
public class ClusterApiControllerIT {

  public static final String TOPIC_NAME = "testtopic";
  public static final String CONSUMER_GROUP = "testconsumergroup";
  public static final String KWCLUSTERAPIUSER = "kwclusterapiuser";
  public static final String AUTHORIZATION = "Authorization";
  public static final String BEARER_PREFIX = "Bearer ";
  public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
  public static final String TEST_MESSAGE = "A test message.";
  public static final String BOOTSTRAP_SERVERS = "localhost:9092";
  public static final String BOOTSTRAP_SERVERS_SSL = "localhost:9093";

  @Autowired private EmbeddedKafkaZKBroker embeddedKafkaBroker;

  @Value("${klaw.clusterapi.access.base64.secret}")
  private String clusterAccessSecret;

  private final UtilMethods utilMethods = new UtilMethods();

  @Autowired private MockMvc mvc;
  ObjectMapper mapper = new ObjectMapper();

  @MockBean SchemaService schemaService;

  @Test
  @Order(1)
  public void getKafkaServerStatus() throws Exception {
    String url =
        "/topics/getStatus/" + BOOTSTRAP_SERVERS + "/PLAINTEXT/DEV1/kafka/kafkaFlavor/Apache Kafka";
    MockHttpServletResponse response =
        mvc.perform(
                MockMvcRequestBuilders.get(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .header(
                        AUTHORIZATION,
                        BEARER_PREFIX
                            + utilMethods.generateToken(KWCLUSTERAPIUSER, clusterAccessSecret, 3L))
                    .accept(MediaType.APPLICATION_JSON))
            .andExpect(status().isOk())
            .andReturn()
            .getResponse();

    ClusterStatus clusterStatus =
        OBJECT_MAPPER.readValue(response.getContentAsString(), ClusterStatus.class);
    assertThat(clusterStatus).isEqualTo(ONLINE);
  }

  @Test
  @Order(2)
  public void getKafkaServerStatusSSL() throws Exception {
    String url =
        "/topics/getStatus/" + BOOTSTRAP_SERVERS_SSL + "/SSL/DEV2/kafka/kafkaFlavor/Apache Kafka";
    MockHttpServletResponse response =
        mvc.perform(
                MockMvcRequestBuilders.get(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .header(
                        AUTHORIZATION,
                        BEARER_PREFIX
                            + utilMethods.generateToken(KWCLUSTERAPIUSER, clusterAccessSecret, 3L))
                    .accept(MediaType.APPLICATION_JSON))
            .andExpect(status().isOk())
            .andReturn()
            .getResponse();

    ClusterStatus clusterStatus =
        OBJECT_MAPPER.readValue(response.getContentAsString(), ClusterStatus.class);
    assertThat(clusterStatus).isEqualTo(ONLINE);
  }

  @Test
  @Order(3)
  public void createTopics() throws Exception {
    String topicName = "testtopic";
    ClusterTopicRequest clusterTopicRequest = createTopicRequest(topicName);
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterTopicRequest);
    String url = "/topics/createTopics";
    executeCreateTopicRequest(jsonReq, url);

    embeddedKafkaBroker.doWithAdmin(
        adminClient -> {
          try {
            Set<String> topicsSet = adminClient.listTopics().names().get();
            assertThat(topicsSet).contains(topicName);
          } catch (InterruptedException | ExecutionException e) {
            log.error("Error : ", e);
          }
        });
  }

  @Test
  @Order(4)
  public void updateTopics() throws Exception {
    String topicName = "testtopic";
    ClusterTopicRequest clusterTopicRequest = updateTopicRequest(topicName);

    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterTopicRequest);
    String url = "/topics/updateTopics";
    mvc.perform(
            MockMvcRequestBuilders.post(url)
                .contentType(MediaType.APPLICATION_JSON)
                .content(jsonReq)
                .header(
                    AUTHORIZATION,
                    BEARER_PREFIX
                        + utilMethods.generateToken(KWCLUSTERAPIUSER, clusterAccessSecret, 3L))
                .accept(MediaType.APPLICATION_JSON))
        .andExpect(status().isOk())
        .andReturn()
        .getResponse();
    Thread.sleep(300);

    embeddedKafkaBroker.doWithAdmin(
        adminClient -> {
          try {
            ConfigResource configResource =
                new ConfigResource(ConfigResource.Type.TOPIC, clusterTopicRequest.getTopicName());
            Config topicConfig =
                adminClient
                    .describeConfigs(Collections.singleton(configResource))
                    .all()
                    .get(10, TimeUnit.SECONDS)
                    .get(configResource);
            assertThat(topicConfig.get("compression.type").value()).isEqualTo("snappy");
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("Error : ", e);
          }
        });
  }

  @Test
  @Order(5)
  public void createTopicsExistingTopicSameConfigSuccess() throws Exception {
    String topicName = "testtopic";
    ClusterTopicRequest clusterTopicRequest = createTopicRequest(topicName);
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterTopicRequest);
    String url = "/topics/createTopics";
    MockHttpServletResponse response = executeCreateTopicRequest(jsonReq, url);
    ApiResponse apiResponse =
        new ObjectMapper().readValue(response.getContentAsString(), new TypeReference<>() {});
    assertThat(apiResponse.isSuccess()).isTrue();
  }

  @Test
  @Order(6)
  public void createTopicsExistingTopicDifferentConfigFailure() throws Exception {
    String topicName = "testtopic";
    ClusterTopicRequest clusterTopicRequest = createTopicRequest(topicName);
    clusterTopicRequest.setReplicationFactor((short) 4);
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterTopicRequest);
    String url = "/topics/createTopics";
    MockHttpServletResponse response = executeCreateTopicRequest(jsonReq, url);
    ApiResponse apiResponse =
        new ObjectMapper().readValue(response.getContentAsString(), new TypeReference<>() {});
    assertThat(apiResponse.getMessage()).contains("TopicExistsException");
  }

  @Test
  @Order(7)
  public void createAclProducerIPAddress() throws Exception {
    String topicName = "testtopic";
    String ipHost = "11.12.13.14";
    ClusterAclRequest clusterAclRequest =
        ClusterAclRequest.builder()
            .clusterName("DEV2")
            .topicName(topicName)
            .env(BOOTSTRAP_SERVERS_SSL)
            .protocol(KafkaSupportedProtocol.SSL)
            .aclIp(ipHost)
            .aclNativeType(AclsNativeType.NATIVE.name())
            .aclIpPrincipleType(AclIPPrincipleType.IP_ADDRESS.name())
            .aclType(AclType.PRODUCER.value)
            .isPrefixAcl(false)
            .requestOperationType(RequestOperationType.CREATE)
            .build();
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterAclRequest);
    String url = "/topics/createAcls";
    executeCreateTopicRequest(jsonReq, url);

    ResourcePatternFilter resourceFilter =
        new ResourcePatternFilter(ResourceType.TOPIC, topicName, PatternType.LITERAL);
    final List<KafkaFuture<Collection<AclBinding>>> aclBindingFutureList = new ArrayList<>();

    Thread.sleep(300);
    embeddedKafkaBroker.doWithAdmin(
        adminClient ->
            aclBindingFutureList.add(
                adminClient
                    .describeAcls(
                        new AclBindingFilter(
                            resourceFilter,
                            new AccessControlEntryFilter(
                                "User:*", ipHost, AclOperation.WRITE, AclPermissionType.ALLOW)))
                    .values()));
    Collection<AclBinding> aclBindings = aclBindingFutureList.get(0).get();

    assertThat(aclBindings.size()).isEqualTo(1);
  }

  @Test
  @Order(8)
  public void createAclConsumerIPAddress() throws Exception {
    String topicName = "testtopic";
    String ipHost = "11.12.13.14";
    String consumerGroup = "testconsumergroup";
    ClusterAclRequest clusterAclRequest =
        ClusterAclRequest.builder()
            .clusterName("DEV2")
            .topicName(topicName)
            .env(BOOTSTRAP_SERVERS_SSL)
            .protocol(KafkaSupportedProtocol.SSL)
            .aclIp(ipHost)
            .aclNativeType(AclsNativeType.NATIVE.name())
            .aclIpPrincipleType(AclIPPrincipleType.IP_ADDRESS.name())
            .aclType(AclType.CONSUMER.value)
            .consumerGroup(consumerGroup)
            .isPrefixAcl(false)
            .requestOperationType(RequestOperationType.CREATE)
            .build();
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterAclRequest);
    String url = "/topics/createAcls";
    executeCreateTopicRequest(jsonReq, url);

    ResourcePatternFilter resourceFilter =
        new ResourcePatternFilter(ResourceType.TOPIC, topicName, PatternType.LITERAL);
    final List<KafkaFuture<Collection<AclBinding>>> aclBindingFutureList = new ArrayList<>();

    // verify topic read access
    Thread.sleep(300);
    embeddedKafkaBroker.doWithAdmin(
        adminClient ->
            aclBindingFutureList.add(
                adminClient
                    .describeAcls(
                        new AclBindingFilter(
                            resourceFilter,
                            new AccessControlEntryFilter(
                                "User:*", ipHost, AclOperation.READ, AclPermissionType.ALLOW)))
                    .values()));
    Collection<AclBinding> aclBindings = aclBindingFutureList.get(0).get();

    assertThat(aclBindings.size()).isEqualTo(1);

    // verify consumer group read access
    ResourcePatternFilter resourceFilter1 =
        new ResourcePatternFilter(ResourceType.GROUP, consumerGroup, PatternType.LITERAL);
    List<KafkaFuture<Collection<AclBinding>>> aclBindingFutureList1 = new ArrayList<>();

    Thread.sleep(300);
    embeddedKafkaBroker.doWithAdmin(
        adminClient ->
            aclBindingFutureList1.add(
                adminClient
                    .describeAcls(
                        new AclBindingFilter(
                            resourceFilter1,
                            new AccessControlEntryFilter(
                                "User:*", ipHost, AclOperation.READ, AclPermissionType.ALLOW)))
                    .values()));
    Collection<AclBinding> aclBindings1 = aclBindingFutureList.get(0).get();

    assertThat(aclBindings1.size()).isEqualTo(1);
  }

  @Test
  @Order(9)
  public void createAclProducerPrincipal() throws Exception {
    String topicName = "testtopic";
    String principle = "CN=host,OU=dept";
    ClusterAclRequest clusterAclRequest =
        ClusterAclRequest.builder()
            .clusterName("DEV2")
            .topicName(topicName)
            .env(BOOTSTRAP_SERVERS_SSL)
            .protocol(KafkaSupportedProtocol.SSL)
            .aclSsl(principle)
            .aclNativeType(AclsNativeType.NATIVE.name())
            .aclIpPrincipleType(AclIPPrincipleType.PRINCIPAL.name())
            .aclType(AclType.PRODUCER.value)
            .isPrefixAcl(false)
            .requestOperationType(RequestOperationType.CREATE)
            .build();
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterAclRequest);
    String url = "/topics/createAcls";
    executeCreateTopicRequest(jsonReq, url);

    ResourcePatternFilter resourceFilter =
        new ResourcePatternFilter(ResourceType.TOPIC, topicName, PatternType.LITERAL);
    final List<KafkaFuture<Collection<AclBinding>>> aclBindingFutureList = new ArrayList<>();

    Thread.sleep(300);
    embeddedKafkaBroker.doWithAdmin(
        adminClient ->
            aclBindingFutureList.add(
                adminClient
                    .describeAcls(
                        new AclBindingFilter(
                            resourceFilter,
                            new AccessControlEntryFilter(
                                "User:" + principle,
                                "*",
                                AclOperation.WRITE,
                                AclPermissionType.ALLOW)))
                    .values()));
    Collection<AclBinding> aclBindings = aclBindingFutureList.get(0).get(3, TimeUnit.SECONDS);

    assertThat(aclBindings.size()).isEqualTo(1);
  }

  @Test
  @Order(10)
  public void createAclConsumerPrincipal() throws Exception {
    String principle = "CN=host,OU=dept";

    ClusterAclRequest clusterAclRequest =
        ClusterAclRequest.builder()
            .clusterName("DEV2")
            .topicName(TOPIC_NAME)
            .env(BOOTSTRAP_SERVERS_SSL)
            .protocol(KafkaSupportedProtocol.SSL)
            .aclSsl(principle)
            .aclNativeType(AclsNativeType.NATIVE.name())
            .aclIpPrincipleType(AclIPPrincipleType.PRINCIPAL.name())
            .aclType(AclType.CONSUMER.value)
            .consumerGroup(CONSUMER_GROUP)
            .isPrefixAcl(false)
            .requestOperationType(RequestOperationType.CREATE)
            .build();
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterAclRequest);
    String url = "/topics/createAcls";
    executeCreateTopicRequest(jsonReq, url);

    ResourcePatternFilter resourceFilter =
        new ResourcePatternFilter(ResourceType.TOPIC, TOPIC_NAME, PatternType.LITERAL);
    final List<KafkaFuture<Collection<AclBinding>>> aclBindingFutureList = new ArrayList<>();

    // verify topic read access
    Thread.sleep(300);
    embeddedKafkaBroker.doWithAdmin(
        adminClient ->
            aclBindingFutureList.add(
                adminClient
                    .describeAcls(
                        new AclBindingFilter(
                            resourceFilter,
                            new AccessControlEntryFilter(
                                "User:" + principle,
                                "*",
                                AclOperation.READ,
                                AclPermissionType.ALLOW)))
                    .values()));
    Collection<AclBinding> aclBindings = aclBindingFutureList.get(0).get();

    assertThat(aclBindings.size()).isEqualTo(1);

    // verify consumer group read access
    ResourcePatternFilter resourceFilter1 =
        new ResourcePatternFilter(ResourceType.GROUP, CONSUMER_GROUP, PatternType.LITERAL);
    List<KafkaFuture<Collection<AclBinding>>> aclBindingFutureList1 = new ArrayList<>();

    Thread.sleep(300);
    embeddedKafkaBroker.doWithAdmin(
        adminClient ->
            aclBindingFutureList1.add(
                adminClient
                    .describeAcls(
                        new AclBindingFilter(
                            resourceFilter1,
                            new AccessControlEntryFilter(
                                "User:" + principle,
                                "*",
                                AclOperation.READ,
                                AclPermissionType.ALLOW)))
                    .values()));
    Collection<AclBinding> aclBindings1 = aclBindingFutureList.get(0).get();

    assertThat(aclBindings1.size()).isEqualTo(1);
  }

  @DisabledOnOs({OS.WINDOWS})
  @Test
  @Order(11)
  public void resetConsumerOffsetsToEarliest() throws Exception {
    produceAndConsumeRecords(true); // produce 10 records and consume all records

    String url = "/topics/consumerGroupOffsets/reset/" + BOOTSTRAP_SERVERS_SSL + "/SSL/" + "DEV2";
    ResetConsumerGroupOffsetsRequest resetConsumerGroupOffsetsRequest =
        ResetConsumerGroupOffsetsRequest.builder()
            .offsetResetType(OffsetResetType.EARLIEST)
            .consumerGroup(CONSUMER_GROUP)
            .topicName(TOPIC_NAME)
            .build();
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(resetConsumerGroupOffsetsRequest);

    String response =
        mvc.perform(
                MockMvcRequestBuilders.post(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .content(jsonReq)
                    .header(
                        AUTHORIZATION,
                        BEARER_PREFIX
                            + utilMethods.generateToken(KWCLUSTERAPIUSER, clusterAccessSecret, 3L))
                    .accept(MediaType.APPLICATION_JSON))
            .andExpect(status().isOk())
            .andReturn()
            .getResponse()
            .getContentAsString();
    ApiResponse apiResponse = new ObjectMapper().readValue(response, ApiResponse.class);

    Map<OffsetsTiming, Map<String, Long>> offsetsResetResponse = (Map) apiResponse.getData();
    assertThat(apiResponse.isSuccess()).isTrue();
    assertThat(offsetsResetResponse).hasSize(2);
    String offsetsBefore =
        offsetsResetResponse
                .get(OffsetsTiming.BEFORE_OFFSET_RESET.getValue())
                .get(TOPIC_NAME + "-0")
            + "";
    String offsetsAfter =
        offsetsResetResponse.get(OffsetsTiming.AFTER_OFFSET_RESET.getValue()).get(TOPIC_NAME + "-0")
            + "";

    assertThat(offsetsBefore).isEqualTo("10");
    String expectedOffsetAfterReset = "0";
    assertThat(offsetsAfter).isEqualTo(expectedOffsetAfterReset);

    // verify consumer group offset position from admin client
    Thread.sleep(300);
    Map<String, Long> currentOffsetPositionsMap = new TreeMap<>();
    embeddedKafkaBroker.doWithAdmin(
        adminClient -> {
          try {
            Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsetAndMetadataMap =
                adminClient
                    .listConsumerGroupOffsets(CONSUMER_GROUP)
                    .partitionsToOffsetAndMetadata()
                    .get();
            for (TopicPartition topicPartition : topicPartitionOffsetAndMetadataMap.keySet()) {
              currentOffsetPositionsMap.put(
                  topicPartition.toString(),
                  topicPartitionOffsetAndMetadataMap.get(topicPartition).offset());
            }
          } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
          }
        });
    assertThat(currentOffsetPositionsMap.get(TOPIC_NAME + "-0") + "")
        .isEqualTo(expectedOffsetAfterReset);
  }

  @DisabledOnOs({OS.WINDOWS})
  @Test
  @Order(12)
  public void resetConsumerOffsetsToLatest() throws Exception {
    produceAndConsumeRecords(true); // produce 10 more records

    String url = "/topics/consumerGroupOffsets/reset/" + BOOTSTRAP_SERVERS_SSL + "/SSL/" + "DEV2";
    ResetConsumerGroupOffsetsRequest resetConsumerGroupOffsetsRequest =
        ResetConsumerGroupOffsetsRequest.builder()
            .offsetResetType(OffsetResetType.LATEST)
            .consumerGroup(CONSUMER_GROUP)
            .topicName(TOPIC_NAME)
            .build();
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(resetConsumerGroupOffsetsRequest);

    String response =
        mvc.perform(
                MockMvcRequestBuilders.post(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .content(jsonReq)
                    .header(
                        AUTHORIZATION,
                        BEARER_PREFIX
                            + utilMethods.generateToken(KWCLUSTERAPIUSER, clusterAccessSecret, 3L))
                    .accept(MediaType.APPLICATION_JSON))
            .andExpect(status().isOk())
            .andReturn()
            .getResponse()
            .getContentAsString();
    ApiResponse apiResponse = new ObjectMapper().readValue(response, ApiResponse.class);

    Map<OffsetsTiming, Map<String, Long>> offsetsResetResponse = (Map) apiResponse.getData();
    assertThat(apiResponse.isSuccess()).isTrue();
    assertThat(offsetsResetResponse).hasSize(2);
    String offsetsBefore =
        offsetsResetResponse
                .get(OffsetsTiming.BEFORE_OFFSET_RESET.getValue())
                .get(TOPIC_NAME + "-0")
            + "";
    String offsetsAfter =
        offsetsResetResponse.get(OffsetsTiming.AFTER_OFFSET_RESET.getValue()).get(TOPIC_NAME + "-0")
            + "";

    String expectedOffsetsBeforeAfter = "20";
    assertThat(offsetsBefore).isEqualTo(expectedOffsetsBeforeAfter);
    assertThat(offsetsAfter).isEqualTo(expectedOffsetsBeforeAfter);

    // verify consumer group offset position from admin client
    Thread.sleep(300);
    Map<String, Long> currentOffsetPositionsMap = new TreeMap<>();
    embeddedKafkaBroker.doWithAdmin(
        adminClient -> {
          try {
            Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsetAndMetadataMap =
                adminClient
                    .listConsumerGroupOffsets(CONSUMER_GROUP)
                    .partitionsToOffsetAndMetadata()
                    .get();
            for (TopicPartition topicPartition : topicPartitionOffsetAndMetadataMap.keySet()) {
              currentOffsetPositionsMap.put(
                  topicPartition.toString(),
                  topicPartitionOffsetAndMetadataMap.get(topicPartition).offset());
            }
          } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
          }
        });
    assertThat(currentOffsetPositionsMap.get(TOPIC_NAME + "-0") + "")
        .isEqualTo(expectedOffsetsBeforeAfter);
  }

  @DisabledOnOs({OS.WINDOWS})
  @Test
  @Order(13)
  public void resetConsumerOffsetsToLatestDontConsumeRecs() throws Exception {
    produceAndConsumeRecords(false); // produce 10 more records

    String url = "/topics/consumerGroupOffsets/reset/" + BOOTSTRAP_SERVERS_SSL + "/SSL/" + "DEV2";
    ResetConsumerGroupOffsetsRequest resetConsumerGroupOffsetsRequest =
        ResetConsumerGroupOffsetsRequest.builder()
            .offsetResetType(OffsetResetType.LATEST)
            .consumerGroup(CONSUMER_GROUP)
            .topicName(TOPIC_NAME)
            .build();
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(resetConsumerGroupOffsetsRequest);

    String response =
        mvc.perform(
                MockMvcRequestBuilders.post(url)
                    .contentType(MediaType.APPLICATION_JSON)
                    .content(jsonReq)
                    .header(
                        AUTHORIZATION,
                        BEARER_PREFIX
                            + utilMethods.generateToken(KWCLUSTERAPIUSER, clusterAccessSecret, 3L))
                    .accept(MediaType.APPLICATION_JSON))
            .andExpect(status().isOk())
            .andReturn()
            .getResponse()
            .getContentAsString();
    ApiResponse apiResponse = new ObjectMapper().readValue(response, ApiResponse.class);

    Map<OffsetsTiming, Map<String, Long>> offsetsResetResponse = (Map) apiResponse.getData();
    assertThat(apiResponse.isSuccess()).isTrue();
    assertThat(offsetsResetResponse).hasSize(2);
    String offsetsBefore =
        offsetsResetResponse
                .get(OffsetsTiming.BEFORE_OFFSET_RESET.getValue())
                .get(TOPIC_NAME + "-0")
            + "";
    String offsetsAfter =
        offsetsResetResponse.get(OffsetsTiming.AFTER_OFFSET_RESET.getValue()).get(TOPIC_NAME + "-0")
            + "";

    String expectedOffsetsBeforeAfter = "20";
    assertThat(offsetsBefore).isEqualTo(expectedOffsetsBeforeAfter);
    assertThat(offsetsAfter).isEqualTo("30");

    // verify consumer group offset position from admin client
    Thread.sleep(300);
    Map<String, Long> currentOffsetPositionsMap = new TreeMap<>();
    embeddedKafkaBroker.doWithAdmin(
        adminClient -> {
          try {
            Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsetAndMetadataMap =
                adminClient
                    .listConsumerGroupOffsets(CONSUMER_GROUP)
                    .partitionsToOffsetAndMetadata()
                    .get();
            for (TopicPartition topicPartition : topicPartitionOffsetAndMetadataMap.keySet()) {
              currentOffsetPositionsMap.put(
                  topicPartition.toString(),
                  topicPartitionOffsetAndMetadataMap.get(topicPartition).offset());
            }
          } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
          }
        });
    assertThat(currentOffsetPositionsMap.get(TOPIC_NAME + "-0") + "").isEqualTo("30");
  }

  @Test
  @Order(14)
  public void deleteTopics() throws Exception {
    // Create a topic
    String topicName = "testtopic-todelete";
    ClusterTopicRequest clusterTopicRequest = createTopicRequest(topicName);
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterTopicRequest);
    String url = "/topics/createTopics";
    executeCreateTopicRequest(jsonReq, url);

    embeddedKafkaBroker.doWithAdmin(
        adminClient -> {
          try {
            Set<String> topicsSet = adminClient.listTopics().names().get();
            assertThat(topicsSet).contains(topicName);
          } catch (InterruptedException | ExecutionException e) {
            log.error("Error : ", e);
          }
        });

    // Delete the topic
    clusterTopicRequest = clusterTopicRequest.toBuilder().deleteAssociatedSchema(true).build();
    jsonReq = OBJECT_MAPPER.writer().writeValueAsString(clusterTopicRequest);
    url = "/topics/deleteTopics";
    when(schemaService.deleteSchema(any()))
        .thenReturn(
            ApiResponse.builder().success(true).message(ApiResultStatus.SUCCESS.value).build());
    MockHttpServletResponse response = executeCreateTopicRequest(jsonReq, url);

    embeddedKafkaBroker.doWithAdmin(
        adminClient -> {
          try {
            Set<String> topicsSet = adminClient.listTopics().names().get();
            assertThat(topicsSet).doesNotContain(topicName);
          } catch (InterruptedException | ExecutionException e) {
            log.error("Error : ", e);
          }
        });

    ApiResponse apiResponse = mapper.readValue(response.getContentAsString(), ApiResponse.class);
    assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @DisabledOnOs({OS.WINDOWS})
  @Test
  @Order(15)
  public void resetOffsetsNonExistingTopic() throws Exception {
    String url = "/topics/consumerGroupOffsets/reset/" + BOOTSTRAP_SERVERS_SSL + "/SSL/" + "DEV2";
    String nonExistingTopic = "topicdoesnotexist";
    ResetConsumerGroupOffsetsRequest resetConsumerGroupOffsetsRequest =
        ResetConsumerGroupOffsetsRequest.builder()
            .offsetResetType(OffsetResetType.LATEST)
            .consumerGroup(CONSUMER_GROUP)
            .topicName(nonExistingTopic)
            .build();
    String jsonReq = OBJECT_MAPPER.writer().writeValueAsString(resetConsumerGroupOffsetsRequest);

    Exception thrown =
        Assertions.assertThrows(
            Exception.class,
            () ->
                mvc.perform(
                        MockMvcRequestBuilders.post(url)
                            .contentType(MediaType.APPLICATION_JSON)
                            .content(jsonReq)
                            .header(
                                AUTHORIZATION,
                                BEARER_PREFIX
                                    + utilMethods.generateToken(
                                        KWCLUSTERAPIUSER, clusterAccessSecret, 3L))
                            .accept(MediaType.APPLICATION_JSON))
                    .andExpect(status().isOk())
                    .andReturn()
                    .getResponse()
                    .getContentAsString());
    assertThat(thrown.getMessage()).contains("Topic " + nonExistingTopic + " does not exist.");
  }

  @DisabledOnOs({OS.WINDOWS})
  @Test
  @Order(16)
  public void getTopicContents() throws Exception {
