package com.assu.study.chap05;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;

public class AdminClientSample2 {

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000); // optional
    props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 1000); // optional
    AdminClient adminClient = AdminClient.create(props);

    // ======= 컨슈머 그룹 조회
    adminClient.listConsumerGroups().valid().get().forEach(System.out::println);
  }
}

package com.assu.study.chap05;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;

@Slf4j
public class AdminClientSample2 {
  private static final String CONSUMER_GROUP = "TestConsumerGroup";
  private static final List<String> CONSUMER_GROUP_LIST = List.of(CONSUMER_GROUP);

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000); // optional
    props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 1000); // optional
    AdminClient adminClient = AdminClient.create(props);

    // ======= 특정 컨슈머 그룹의 상세 정보 조회
    ConsumerGroupDescription groupDescription =
        adminClient
            .describeConsumerGroups(CONSUMER_GROUP_LIST) // DescribeConsumerGroupsResult 반환
            .describedGroups() // Map<String, KafkaFuture<ConsumerGroupDescription>> 반환
            .get(CONSUMER_GROUP) // KafkaFuture<ConsumerGroupDescription> 반환
            .get();
    
    log.info("Description of Consumer group: {} - {}", CONSUMER_GROUP, groupDescription);
  }
}

package com.assu.study.chap05;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

@Slf4j
public class AdminClientSample2 {
  private static final String CONSUMER_GROUP = "TestConsumerGroup";
  private static final List<String> CONSUMER_GROUP_LIST = List.of(CONSUMER_GROUP);

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000); // optional
    props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 1000); // optional
    AdminClient adminClient = AdminClient.create(props);
    
    // ======= 컨슈머 그룹에서 오프셋 커밋 정보 조회
    // 1)
    // 컨슈머 그룹이 사용 중인 모든 토픽 파티션을 key 로, 각각의 토픽 파티션에 대해 마지막으로 커밋된 오프셋을 value 로 하는 맵 조회
    // `describeConsumerGroups()` 와 달리 `listConsumerGroupOffsets()` 은 컨슈머 그룹의 모음이 아닌 하나의 컨슈머 그룹을 받음
    Map<TopicPartition, OffsetAndMetadata> offsets =
        adminClient.listConsumerGroupOffsets(CONSUMER_GROUP).partitionsToOffsetAndMetadata().get();

    Map<TopicPartition, OffsetSpec> reqLatestOffsets = new HashMap<>();
    Map<TopicPartition, OffsetSpec> reqEarliestOffsets = new HashMap<>();
    Map<TopicPartition, OffsetSpec> reqOlderOffsets = new HashMap<>();

    Instant resetTo = ZonedDateTime.now(ZoneId.of("Asia/Seoul")).minusHours(2).toInstant();

    // 2)
    // 컨슈머 그룹에서 커밋한 토픽의 모든 파티션에 대해 최신 오프셋, 가장 오래된 오프셋, 2시간 전의 오프셋 조회
    for (TopicPartition tp : offsets.keySet()) {
      reqLatestOffsets.put(tp, OffsetSpec.latest());
      reqEarliestOffsets.put(tp, OffsetSpec.earliest());
      reqOlderOffsets.put(tp, OffsetSpec.forTimestamp(resetTo.toEpochMilli()));
    }

    Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> latestOffsets =
        adminClient.listOffsets(reqLatestOffsets).all().get();
    //    Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> earliestOffsets =
    //        adminClient.listOffsets(reqEarliestOffsets).all().get();
    //    Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> orderOffsets =
    //        adminClient.listOffsets(reqOlderOffsets).all().get();

    // 3)
    // 모든 파티션을 반복해서 각각의 파티션에 대해 마지막으로 커밋된 오프셋, 파티션의 마지막 오프셋, 둘 사이의 lag 출력
    for (Map.Entry<TopicPartition, OffsetAndMetadata> e : offsets.entrySet()) {
      String topic = e.getKey().topic();
      int partition = e.getKey().partition();
      long committedOffset = e.getValue().offset();

      long latestOffset = latestOffsets.get(e.getKey()).offset();
      // 아래는 확실하지 않음
      //      long earliestOffset = earliestOffsets.get(e.getKey()).offset();
      //      long orderOffset = orderOffsets.get(e.getKey()).offset();

      log.info(
          "Consumer group {} has committed offset {} to topic {}, partition {}.",
          CONSUMER_GROUP,
          committedOffset,
          topic,
          partition);
      log.info(
          "The latest offset in the partition is {} to consumer group is {} records behind.",
          latestOffset,
          (latestOffset - committedOffset));
      // 아래는 확실하지 않음
      // log.info("The earliest offset: {}, order offset: {}", earliestOffset, orderOffset);
    }
  }
}

package com.assu.study.chap05;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.UnknownMemberIdException;

@Slf4j
public class AdminClientSample2 {
  private static final String CONSUMER_GROUP = "TestConsumerGroup";
  private static final List<String> CONSUMER_GROUP_LIST = List.of(CONSUMER_GROUP);

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000); // optional
    props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 1000); // optional
    AdminClient adminClient = AdminClient.create(props);

    // ======= 컨슈머 그룹에서 오프셋 커밋 정보 조회
    Map<TopicPartition, OffsetAndMetadata> offsets =
        adminClient.listConsumerGroupOffsets(CONSUMER_GROUP).partitionsToOffsetAndMetadata().get();

    Map<TopicPartition, OffsetSpec> reqLatestOffsets = new HashMap<>();
    Map<TopicPartition, OffsetSpec> reqEarliestOffsets = new HashMap<>();
    Map<TopicPartition, OffsetSpec> reqOlderOffsets = new HashMap<>();

    for (TopicPartition tp : offsets.keySet()) {
      reqLatestOffsets.put(tp, OffsetSpec.latest());
      reqEarliestOffsets.put(tp, OffsetSpec.earliest());
      reqOlderOffsets.put(tp, OffsetSpec.forTimestamp(resetTo.toEpochMilli()));
    }

    // ...
    
    // ======= 컨슈머 그룹을 특정 오프셋으로 리셋

    // 1)
    // 맨 앞 오프셋부터 처리하도록 컨슈머 그룹을 리셋하기 위해 토픽의 맨 앞 오프셋 값 조회
    // (reqOlderOffsets 를 사용하면 2시간 전으로 오프셋 리셋 가능)
    Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> earliestOffsets =
        adminClient.listOffsets(reqEarliestOffsets).all().get();

    Map<TopicPartition, OffsetAndMetadata> resetOffsets = new HashMap<>();

    // 2) 
    // 반복문을 이용하여 listOffsets() 가 리턴한 ListOffsetsResultInfo 의 맵 객체를 alterConsumerGroupOffsets() 를 
    // 호출하는데 필요한 OffsetAndMetadata 의 맵 객체로 변환
    for (Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> e :
        earliestOffsets.entrySet()) {
      log.info("Will reset topic-partition: {}, to offset: {}", e.getKey(), e.getValue().offset());
      resetOffsets.put(e.getKey(), new OffsetAndMetadata(e.getValue().offset()));
    }

    try {
      // 3) 
      //  alterConsumerGroupOffsets() 를 호출한 뒤 get() 을 호출하여 Future 객체가 작업을 성공적으로 완료할 때까지 기다림
      adminClient.alterConsumerGroupOffsets(CONSUMER_GROUP, resetOffsets).all().get();
    } catch (ExecutionException e) {
      log.error(
          "Failed to update the offsets committed by group: {}, with error: {}",
          CONSUMER_GROUP,
          e.getMessage());

      // 4)
      // alterConsumerGroupOffsets() 가 실패하는 가장 흔한 이유 중 하나는 컨슈머 그룹을 미리 정지시키지 않아서임
      // 특정 컨슈머 그룹을 정지시키는 어드민 명령은 없기 때문에 컨슈머 애플리케이션을 정지시키는 것 외에는 방법이 없음
      // 만일 컨슈머 그룹이 여전히 돌아가고 있는 중이라면, 컨슈머 코디네이터 입장에서는 컨슈머 그룹에 대한 오프셋 변경 시도를 그룹의 멤버가 아닌 
      // 클라이언트가 오프셋을 커밋하려는 것으로 간주함
      // 이 경우 UnknownMemberIdException 이 발생됨
      if (e.getCause() instanceof UnknownMemberIdException) {
        log.error("Check if consumer group is still active.");
      }
    }
  }
}

DescribeClusterResult cluster = adminClient.describeCluster();

log.info("Connected to cluster: {}", cluster.clusterId().get());
log.info("The brokers in the cluster are: ");
cluster.nodes().get().forEach(node -> log.info("   * {}", node));
log.info("The controller is {}", cluster.controller().get());

package com.assu.study.chap05;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.common.errors.InvalidPartitionsException;

@Slf4j
public class AdminClientSample3 {
  public static final String TOPIC_NAME = "sample-topic";
  public static final int NUM_PARTITIONS = 6;

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000); // optional
    props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 1000); // optional
    AdminClient adminClient = AdminClient.create(props);

    // ======= 토픽에 파티션 추가
    Map<String, NewPartitions> newPartitions = new HashMap<>();
    
    // 토픽의 파티션을 확장할 때는 새로 추가될 파티션 수가 아닌 파티션이 추가된 뒤의 파티션 수를 지정해야 함
    newPartitions.put(TOPIC_NAME, NewPartitions.increaseTo(NUM_PARTITIONS+2));

    try {
      adminClient.createPartitions(newPartitions).all().get();
    } catch (ExecutionException e) {
      if (e.getCause() instanceof InvalidPartitionsException) {
        log.error("Couldn't modify number of partitions in topic: {}", e.getMessage());
      }
    }

  }
}

package com.assu.study.chap05;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.InvalidPartitionsException;
import org.apache.kafka.common.internals.Topic;

@Slf4j
public class AdminClientSample3 {
  public static final String CONSUMER_GROUP = "testConsumerGroup";
  public static final String TOPIC_NAME = "sample-topic";
  public static final int NUM_PARTITIONS = 6;

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000); // optional
    props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 1000); // optional
    AdminClient adminClient = AdminClient.create(props);

    // ======= 토픽에 파티션 추가
    // ...

    // ======= 토픽에서 레코드 삭제
    Map<TopicPartition, OffsetAndMetadata> offsets =
        adminClient.listConsumerGroupOffsets(CONSUMER_GROUP).partitionsToOffsetAndMetadata().get();

    Map<TopicPartition, OffsetSpec> reqOrderOffsets = new HashMap<>();
    Instant resetTo = ZonedDateTime.now(ZoneId.of("Asia/Seoul")).minusHours(2).toInstant();

    for (TopicPartition tp : offsets.keySet()) {
      reqOrderOffsets.put(tp, OffsetSpec.forTimestamp(resetTo.toEpochMilli()));
    }

    Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> olderOffsets =
        adminClient.listOffsets(reqOrderOffsets).all().get();

    Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
    for (Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> e :
        olderOffsets.entrySet()) {
      recordsToDelete.put(e.getKey(), RecordsToDelete.beforeOffset(e.getValue().offset()));
    }
    
    adminClient.deleteRecords(recordsToDelete).all().get();
  }
}

package com.assu.study.chap05;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.*;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.ElectionType;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ElectionNotNeededException;
import org.apache.kafka.common.errors.InvalidPartitionsException;

@Slf4j
public class AdminClientSample3 {
  public static final String CONSUMER_GROUP = "testConsumerGroup";
  public static final String TOPIC_NAME = "sample-topic";
  public static final int NUM_PARTITIONS = 6;

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000); // optional
    props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 1000); // optional
    AdminClient adminClient = AdminClient.create(props);

    // ...

    // ======= 리더 선출
    Set<TopicPartition> electableTopics = new HashSet<>();
    electableTopics.add(new TopicPartition(TOPIC_NAME, 0));

    try {
      // 1)
      // 특정 토픽의 한 파티션에 대해 선호 리더 선출
      // 지정할 수 있는 토픽과 파티션 수에는 제한이 업음
      // 만일 파티션 모음이 아닌 null 값을 지정하여 아래 명령어를 실행할 경우 모든 파티션에 대해 지정된 선출 유형 작업을 시작함
      adminClient.electLeaders(ElectionType.PREFERRED, electableTopics).all().get();
    } catch (ExecutionException e) {
      // 2)
      // 클러스터 상태가 좋다면 아무런 작업도 일어나지 않을 것임
      // 선호 리더 선출과 언클린 리더 선출은 선호 리더가 아닌 replica 가 현재 리더를 맡고 있을 경우에만 수행됨
      if (e.getCause() instanceof ElectionNotNeededException) {
        log.error("All leaders are preferred leaders, no need to do anything.");
      }
    }
  }
}

package com.assu.study.chap05;

import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.*;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.ElectionType;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ElectionNotNeededException;
import org.apache.kafka.common.errors.InvalidPartitionsException;
import org.apache.kafka.common.errors.NoReassignmentInProgressException;

@Slf4j
public class AdminClientSample3 {
  public static final String CONSUMER_GROUP = "testConsumerGroup";
  public static final String TOPIC_NAME = "sample-topic";
  public static final int NUM_PARTITIONS = 6;
  private static final List<String> TOPIC_LIST = List.of(TOPIC_NAME);

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000); // optional
    props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 1000); // optional
    AdminClient adminClient = AdminClient.create(props);

    // ...

    // ======= 새로운 브로커로 파티션 재할당 (replica 재할당)
    Map<TopicPartition, Optional<NewPartitionReassignment>> reassignment = new HashMap<>();

    // 1)
    // 파티션 0 에 새로운 replica 를 추가하고, 새 replica 를 ID 가 1인 새 브로커에 배치
    // 단, 리더는 변경하지 않음
    reassignment.put(
        new TopicPartition(TOPIC_NAME, 0),
        Optional.of(new NewPartitionReassignment(Arrays.asList(0, 1))));

    // 2)
    // 파티션 1 에는 아무런 replica 도 추가하지 않음
    // 단지 이미 있던 replica 를 새 브로커로 옮겼을 뿐임
    // replica 가 하나뿐인만큼 이것이 리더가 됨
    reassignment.put(
        new TopicPartition(TOPIC_NAME, 1),
        Optional.of(new NewPartitionReassignment(Arrays.asList(0))));

    // 3)
    // 파티션 2 에 새로운 replica 를 추가하고 이것을 선호 리더로 설정
    // 다음 선호 리더 선출 시 새로운 브로커에 있는 새로운 replica 로 리더가 바뀌게 됨
    // 이전 replica 는 팔로워가 될 것임
    reassignment.put(
        new TopicPartition(TOPIC_NAME, 2),
        Optional.of(new NewPartitionReassignment(Arrays.asList(1, 0))));

    // 4)
    // 파티션 3 에 대해서는 진행중인 재할당 작업이 업음
    // 하지만 그런게 있다면 작업이 취소되고 재할당 작업이 시작되기 전 상태로 원상복구될 것임
    reassignment.put(new TopicPartition(TOPIC_NAME, 3), Optional.empty());

    try {
      adminClient.alterPartitionReassignments(reassignment).all().get();
    } catch (ExecutionException e) {
      if (e.getCause() instanceof NoReassignmentInProgressException) {
        log.error(
            "We tried cancelling a reassignment that was not happening anyway. Let's ignore this.");
      }
    }

    // 5)
    // 현재 진행중인 재할당을 보여줌
    log.info(
        "Currently reassigning: {}",
        adminClient.listPartitionReassignments().reassignments().get());

    DescribeTopicsResult sampleTopic = adminClient.describeTopics(TOPIC_LIST);
    TopicDescription topicDescription = sampleTopic.topicNameValues().get(TOPIC_NAME).get();

    // 6)
    // 새로운 상태를 보여줌
    // 단, 일관적인 결과가 보일 때까지는 잠시 시간이 걸릴 수 있음
    log.info("Description of sample topic: {}", topicDescription);

    adminClient.close(Duration.ofSeconds(30));
  }
}

package com.assu.study.chap05;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AlterConfigOp;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;

@RequiredArgsConstructor
public class TopicCreator {
  private final AdminClient adminClient;

  // 토픽 이름이 test 로 시작할 경우 토픽을 생성하는 메서드 (좋은 예시는 아님)
  public void maybeCreateTopic(String topicName) throws ExecutionException, InterruptedException {
    Collection<NewTopic> topics = new ArrayList<>();

    // 파티션 1개, replica 1개인 토픽 생성
    topics.add(new NewTopic(topicName, 1, (short) 1));

    if (topicName.toLowerCase().startsWith("test")) {
      adminClient.createTopics(topics);

      // 설정 변경
      ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
      ConfigEntry compaction =
          new ConfigEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);

      Collection<AlterConfigOp> configOp = new ArrayList<>();
      configOp.add(new AlterConfigOp(compaction, AlterConfigOp.OpType.SET));

      Map<ConfigResource, Collection<AlterConfigOp>> alterConfigs = new HashMap<>();
      alterConfigs.put(configResource, configOp);

      adminClient.incrementalAlterConfigs(alterConfigs).all().get();
    }
  }
}

package com.assu.study.chap05;

import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;

import java.util.List;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AlterConfigsResult;
import org.apache.kafka.clients.admin.MockAdminClient;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.junit.jupiter.api.BeforeEach;

// mock 을 사용하여 TopicCreator 테스트
class TopicCreatorTest {
  private AdminClient adminClient;

  @BeforeEach
  void setUp() {
    // id 가 0 인 브로커 생성
    Node broker = new Node(0, "localhost", 9092);

    // 브로커 목록과 컨트롤러를 지정하여 MockAdminClient 객체 생성 (여기서는 하나만 지정함)
    // 나중에 TopicCreator 가 제대로 실행되었는지 확인하기 위해 spy() 메서드의 주입 기능을 사용할 것임
    this.adminClient = spy(new MockAdminClient(List.of(broker), broker));

    // 아래 내용이 없으면 테스트 시
    // `java.lang.UnsupportedOperationException: Not implemented yet` 예외 발생됨
    // --> 없애고 해봤는데 오류 안남

    // Mockito 의 doReturn() 메서드를 사용하여 mock-up 클라이언트가 예외를 발생시키지 않도록 함
    // 테스트하고자 하는 메서드는 AlterConfigResult 객체가 필요하고, 
    // 이 객체는 KafkaFuture 객체를 리턴하는 all() 메서드가 있어야 함
    // 가짜 incrementalAlterConfigs() 메서드는 정확히 이것을 리턴해야 함
    
//    AlterConfigsResult emptyResult = mock(AlterConfigsResult.class);
//    doReturn(KafkaFuture.completedFuture(null)).when(emptyResult).all();
//    doReturn(emptyResult).when(adminClient).incrementalAlterConfigs(any());
  }
}

package com.assu.study.chap05;

import static org.mockito.Mockito.*;

import java.util.List;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AlterConfigsResult;
import org.apache.kafka.clients.admin.MockAdminClient;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

// mock 을 사용하여 TopicCreator 테스트
class TopicCreatorTest {
  private AdminClient adminClient;

  // ...

  @Test
  public void testCreateTopic() throws ExecutionException, InterruptedException {
    TopicCreator tc = new TopicCreator(adminClient);
    tc.maybeCreateTopic("test.is.a.test.topic");

    // createTopics() 가 한 번 호출되었는지 확인
    verify(adminClient, times(1)).createTopics(any());
  }

  @Test
  public void tetNotTopic() throws ExecutionException, InterruptedException {
    TopicCreator tc = new TopicCreator(adminClient);
    tc.maybeCreateTopic("not.a.test");

    // 토픽 이름이 test 로 시작하지 않을 경우 createTopics() 가 한 번도 호출되지 않았는지 확인
    verify(adminClient, never()).createTopics(any());
  }
}